[jira] [Commented] (FLINK-10791) Provide end-to-end test for Kafka 0.11 connector
[ https://issues.apache.org/jira/browse/FLINK-10791?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692802#comment-16692802 ] ASF GitHub Bot commented on FLINK-10791: yanghua commented on issue #7038: [FLINK-10791] Provide end-to-end test for Kafka 0.11 connector URL: https://github.com/apache/flink/pull/7038#issuecomment-440177013 @pnowojski How about providing a `flink-examples/flink-examples-e2e` module to place these end-to-end examples? This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Provide end-to-end test for Kafka 0.11 connector > > > Key: FLINK-10791 > URL: https://issues.apache.org/jira/browse/FLINK-10791 > Project: Flink > Issue Type: Test > Components: E2E Tests >Reporter: vinoyang >Assignee: vinoyang >Priority: Major > Labels: pull-request-available > Fix For: 1.7.0 > > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[GitHub] yanghua commented on issue #7038: [FLINK-10791] Provide end-to-end test for Kafka 0.11 connector
yanghua commented on issue #7038: [FLINK-10791] Provide end-to-end test for Kafka 0.11 connector URL: https://github.com/apache/flink/pull/7038#issuecomment-440177013 @pnowojski How about providing a `flink-examples/flink-examples-e2e` module to place these end-to-end examples? This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-10791) Provide end-to-end test for Kafka 0.11 connector
[ https://issues.apache.org/jira/browse/FLINK-10791?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692786#comment-16692786 ] ASF GitHub Bot commented on FLINK-10791: pnowojski commented on issue #7038: [FLINK-10791] Provide end-to-end test for Kafka 0.11 connector URL: https://github.com/apache/flink/pull/7038#issuecomment-440173975 @tillrohrmann yes, they do not add much value and yes, they are not worth blowing up the release binaries. However they have some value and especially that we are getting and maintaining them for free (as part of end-to-end tests) it would be nice to have them. Flink generally I think lacks of more code examples. Maybe the solution is to come up with some structure that will allow us to ship those (and for other connectors/features?) examples as pure code examples? This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Provide end-to-end test for Kafka 0.11 connector > > > Key: FLINK-10791 > URL: https://issues.apache.org/jira/browse/FLINK-10791 > Project: Flink > Issue Type: Test > Components: E2E Tests >Reporter: vinoyang >Assignee: vinoyang >Priority: Major > Labels: pull-request-available > Fix For: 1.7.0 > > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[GitHub] pnowojski commented on issue #7038: [FLINK-10791] Provide end-to-end test for Kafka 0.11 connector
pnowojski commented on issue #7038: [FLINK-10791] Provide end-to-end test for Kafka 0.11 connector URL: https://github.com/apache/flink/pull/7038#issuecomment-440173975 @tillrohrmann yes, they do not add much value and yes, they are not worth blowing up the release binaries. However they have some value and especially that we are getting and maintaining them for free (as part of end-to-end tests) it would be nice to have them. Flink generally I think lacks of more code examples. Maybe the solution is to come up with some structure that will allow us to ship those (and for other connectors/features?) examples as pure code examples? This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-7243) Add ParquetInputFormat
[
https://issues.apache.org/jira/browse/FLINK-7243?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692776#comment-16692776
]
ASF GitHub Bot commented on FLINK-7243:
---
HuangZhenQiu commented on a change in pull request #6483:
[FLINK-7243][flink-formats] Add parquet input format
URL: https://github.com/apache/flink/pull/6483#discussion_r234892484
##
File path:
flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetRowInputFormat.java
##
@@ -0,0 +1,87 @@
+/*
+ * 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.formats.parquet;
+
+import org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.types.Row;
+
+import org.apache.parquet.schema.MessageType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Timestamp;
+
+
+/**
+ * A subclass of {@link ParquetInputFormat} to read from Parquet files and
convert to {@link Row}.
+ * It is mainly used to integrate with table API and batch SQL.
+ */
+public class ParquetRowInputFormat extends ParquetInputFormat implements
ResultTypeQueryable {
+ private static final long serialVersionUID = 11L;
+ private static final Logger LOG =
LoggerFactory.getLogger(ParquetRowInputFormat.class);
+ private boolean timeStampRewrite;
+ private RowTypeInfo returnType;
+ private int tsIndex;
+
+ public ParquetRowInputFormat(Path path, MessageType messageType) {
+ super(path, messageType);
+ this.returnType = new RowTypeInfo(getFieldTypes(),
getFieldNames());
+ this.timeStampRewrite = false;
+ }
+
+ @Override
+ public TypeInformation getProducedType() {
+ return new RowTypeInfo(getFieldTypes(), getFieldNames());
Review comment:
Yes, you are right.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Add ParquetInputFormat
> --
>
> Key: FLINK-7243
> URL: https://issues.apache.org/jira/browse/FLINK-7243
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Reporter: godfrey he
>Assignee: Zhenqiu Huang
>Priority: Major
> Labels: pull-request-available
>
> Add a {{ParquetInputFormat}} to read data from a Apache Parquet file.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[GitHub] HuangZhenQiu commented on a change in pull request #6483: [FLINK-7243][flink-formats] Add parquet input format
HuangZhenQiu commented on a change in pull request #6483:
[FLINK-7243][flink-formats] Add parquet input format
URL: https://github.com/apache/flink/pull/6483#discussion_r234892484
##
File path:
flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetRowInputFormat.java
##
@@ -0,0 +1,87 @@
+/*
+ * 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.formats.parquet;
+
+import org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.types.Row;
+
+import org.apache.parquet.schema.MessageType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Timestamp;
+
+
+/**
+ * A subclass of {@link ParquetInputFormat} to read from Parquet files and
convert to {@link Row}.
+ * It is mainly used to integrate with table API and batch SQL.
+ */
+public class ParquetRowInputFormat extends ParquetInputFormat implements
ResultTypeQueryable {
+ private static final long serialVersionUID = 11L;
+ private static final Logger LOG =
LoggerFactory.getLogger(ParquetRowInputFormat.class);
+ private boolean timeStampRewrite;
+ private RowTypeInfo returnType;
+ private int tsIndex;
+
+ public ParquetRowInputFormat(Path path, MessageType messageType) {
+ super(path, messageType);
+ this.returnType = new RowTypeInfo(getFieldTypes(),
getFieldNames());
+ this.timeStampRewrite = false;
+ }
+
+ @Override
+ public TypeInformation getProducedType() {
+ return new RowTypeInfo(getFieldTypes(), getFieldNames());
Review comment:
Yes, you are right.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[jira] [Commented] (FLINK-7243) Add ParquetInputFormat
[
https://issues.apache.org/jira/browse/FLINK-7243?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692772#comment-16692772
]
ASF GitHub Bot commented on FLINK-7243:
---
HuangZhenQiu commented on a change in pull request #6483:
[FLINK-7243][flink-formats] Add parquet input format
URL: https://github.com/apache/flink/pull/6483#discussion_r234892171
##
File path:
flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetRowInputFormat.java
##
@@ -0,0 +1,87 @@
+/*
+ * 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.formats.parquet;
+
+import org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.types.Row;
+
+import org.apache.parquet.schema.MessageType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Timestamp;
+
+
+/**
+ * A subclass of {@link ParquetInputFormat} to read from Parquet files and
convert to {@link Row}.
+ * It is mainly used to integrate with table API and batch SQL.
+ */
+public class ParquetRowInputFormat extends ParquetInputFormat implements
ResultTypeQueryable {
+ private static final long serialVersionUID = 11L;
+ private static final Logger LOG =
LoggerFactory.getLogger(ParquetRowInputFormat.class);
+ private boolean timeStampRewrite;
+ private RowTypeInfo returnType;
+ private int tsIndex;
+
+ public ParquetRowInputFormat(Path path, MessageType messageType) {
+ super(path, messageType);
+ this.returnType = new RowTypeInfo(getFieldTypes(),
getFieldNames());
+ this.timeStampRewrite = false;
+ }
+
+ @Override
+ public TypeInformation getProducedType() {
+ return new RowTypeInfo(getFieldTypes(), getFieldNames());
+ }
+
+ @Override
+ protected Row convert(Row row) {
+ if (timeStampRewrite) {
+ row.setField(tsIndex, new Timestamp((long)
row.getField(tsIndex)));
Review comment:
Agree. When I work on HiveTableSource internally, it is 1 year ago on Flink
1.4. As I remember, when I use window functions for group by, for example
TUMBLE(time_attr, interval). The attribute time_attr has to be the type of
Timestamp. It means we need to convert to the field to Timestamp type
somewhere. Do you have any preference in the implementation? Or you just want
to leave it in ParquetTableSource?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Add ParquetInputFormat
> --
>
> Key: FLINK-7243
> URL: https://issues.apache.org/jira/browse/FLINK-7243
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Reporter: godfrey he
>Assignee: Zhenqiu Huang
>Priority: Major
> Labels: pull-request-available
>
> Add a {{ParquetInputFormat}} to read data from a Apache Parquet file.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-10942) Deduplicate common codes in OutputEmitterTest
[
https://issues.apache.org/jira/browse/FLINK-10942?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692770#comment-16692770
]
ASF GitHub Bot commented on FLINK-10942:
zhijiangW commented on issue #7146: [FLINK-10942][network,test] Deduplicate
common codes in OutputEmitterTest
URL: https://github.com/apache/flink/pull/7146#issuecomment-440170847
cc @pnowojski
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Deduplicate common codes in OutputEmitterTest
> -
>
> Key: FLINK-10942
> URL: https://issues.apache.org/jira/browse/FLINK-10942
> Project: Flink
> Issue Type: Sub-task
> Components: Network, Tests
>Reporter: zhijiang
>Assignee: zhijiang
>Priority: Minor
> Labels: pull-request-available
> Fix For: 1.7.1
>
>
> There are many duplicated codes in {{OutputEmitterTest}} to make it difficult
> to maintain. So it is necessary to abstract the common codes to make it
> simple which brings benefits for the following refactor work in
> {{ChannelSelector}} interface.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-8159) Pattern(Flat)SelectFunctions should support RichFunction interface
[
https://issues.apache.org/jira/browse/FLINK-8159?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692768#comment-16692768
]
ASF GitHub Bot commented on FLINK-8159:
---
dianfu commented on issue #5080: [FLINK-8159] [cep] Add rich support for
SelectWrapper and FlatSelectWrapper
URL: https://github.com/apache/flink/pull/5080#issuecomment-440170762
Close this PR as it is duplicate with #7110
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Pattern(Flat)SelectFunctions should support RichFunction interface
> --
>
> Key: FLINK-8159
> URL: https://issues.apache.org/jira/browse/FLINK-8159
> Project: Flink
> Issue Type: Sub-task
> Components: CEP
>Reporter: Dian Fu
>Assignee: Dian Fu
>Priority: Major
> Labels: pull-request-available
> Fix For: 1.8.0
>
>
> {{SelectWrapper}} and {{FlatSelectWrapper}} should extends
> {{AbstractRichFucntion}} and process properly if the underlying functions
> extend RichFunction.
> Things to be very careful about:
> * backwards compatibility (we previously serialized conditions) - changes to
> those interfaces have to be done carefully
> * we want to be able to add dynamic patterns in the future, so at some point
> we have to open also on control message arrival
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[GitHub] HuangZhenQiu commented on a change in pull request #6483: [FLINK-7243][flink-formats] Add parquet input format
HuangZhenQiu commented on a change in pull request #6483:
[FLINK-7243][flink-formats] Add parquet input format
URL: https://github.com/apache/flink/pull/6483#discussion_r234892171
##
File path:
flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetRowInputFormat.java
##
@@ -0,0 +1,87 @@
+/*
+ * 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.formats.parquet;
+
+import org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.types.Row;
+
+import org.apache.parquet.schema.MessageType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Timestamp;
+
+
+/**
+ * A subclass of {@link ParquetInputFormat} to read from Parquet files and
convert to {@link Row}.
+ * It is mainly used to integrate with table API and batch SQL.
+ */
+public class ParquetRowInputFormat extends ParquetInputFormat implements
ResultTypeQueryable {
+ private static final long serialVersionUID = 11L;
+ private static final Logger LOG =
LoggerFactory.getLogger(ParquetRowInputFormat.class);
+ private boolean timeStampRewrite;
+ private RowTypeInfo returnType;
+ private int tsIndex;
+
+ public ParquetRowInputFormat(Path path, MessageType messageType) {
+ super(path, messageType);
+ this.returnType = new RowTypeInfo(getFieldTypes(),
getFieldNames());
+ this.timeStampRewrite = false;
+ }
+
+ @Override
+ public TypeInformation getProducedType() {
+ return new RowTypeInfo(getFieldTypes(), getFieldNames());
+ }
+
+ @Override
+ protected Row convert(Row row) {
+ if (timeStampRewrite) {
+ row.setField(tsIndex, new Timestamp((long)
row.getField(tsIndex)));
Review comment:
Agree. When I work on HiveTableSource internally, it is 1 year ago on Flink
1.4. As I remember, when I use window functions for group by, for example
TUMBLE(time_attr, interval). The attribute time_attr has to be the type of
Timestamp. It means we need to convert to the field to Timestamp type
somewhere. Do you have any preference in the implementation? Or you just want
to leave it in ParquetTableSource?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] zhijiangW commented on issue #7146: [FLINK-10942][network, test] Deduplicate common codes in OutputEmitterTest
zhijiangW commented on issue #7146: [FLINK-10942][network,test] Deduplicate common codes in OutputEmitterTest URL: https://github.com/apache/flink/pull/7146#issuecomment-440170847 cc @pnowojski This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] dianfu commented on issue #5080: [FLINK-8159] [cep] Add rich support for SelectWrapper and FlatSelectWrapper
dianfu commented on issue #5080: [FLINK-8159] [cep] Add rich support for SelectWrapper and FlatSelectWrapper URL: https://github.com/apache/flink/pull/5080#issuecomment-440170762 Close this PR as it is duplicate with #7110 This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-8159) Pattern(Flat)SelectFunctions should support RichFunction interface
[
https://issues.apache.org/jira/browse/FLINK-8159?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692764#comment-16692764
]
ASF GitHub Bot commented on FLINK-8159:
---
dianfu closed pull request #5080: [FLINK-8159] [cep] Add rich support for
SelectWrapper and FlatSelectWrapper
URL: https://github.com/apache/flink/pull/5080
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/FlatSelectTimeoutCepOperator.java
b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/FlatSelectTimeoutCepOperator.java
index 4423bb1dd40..95225271b9e 100644
---
a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/FlatSelectTimeoutCepOperator.java
+++
b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/FlatSelectTimeoutCepOperator.java
@@ -20,7 +20,8 @@
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.api.common.functions.AbstractRichFunction;
+import org.apache.flink.api.common.functions.RichFunction;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.cep.EventComparator;
@@ -28,6 +29,7 @@
import org.apache.flink.cep.PatternFlatTimeoutFunction;
import org.apache.flink.cep.nfa.AfterMatchSkipStrategy;
import org.apache.flink.cep.nfa.compiler.NFACompiler;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.operators.TimestampedCollector;
import org.apache.flink.util.OutputTag;
@@ -102,7 +104,7 @@ protected void processTimedOutSequences(
* in one udf.
*/
@Internal
- public static class FlatSelectWrapper implements
Function {
+ public static class FlatSelectWrapper extends
AbstractRichFunction {
private static final long serialVersionUID =
-8320546120157150202L;
@@ -125,5 +127,25 @@ public FlatSelectWrapper(
this.flatSelectFunction = flatSelectFunction;
this.flatTimeoutFunction = flatTimeoutFunction;
}
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ if (flatSelectFunction instanceof RichFunction) {
+ ((RichFunction)
flatSelectFunction).open(parameters);
+ }
+ if (flatTimeoutFunction instanceof RichFunction) {
+ ((RichFunction)
flatTimeoutFunction).open(parameters);
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (flatSelectFunction instanceof RichFunction) {
+ ((RichFunction) flatSelectFunction).close();
+ }
+ if (flatTimeoutFunction instanceof RichFunction) {
+ ((RichFunction) flatTimeoutFunction).close();
+ }
+ }
}
}
diff --git
a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/SelectTimeoutCepOperator.java
b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/SelectTimeoutCepOperator.java
index cb233a486ec..18a1454cf4c 100644
---
a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/SelectTimeoutCepOperator.java
+++
b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/SelectTimeoutCepOperator.java
@@ -19,7 +19,8 @@
package org.apache.flink.cep.operator;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.api.common.functions.AbstractRichFunction;
+import org.apache.flink.api.common.functions.RichFunction;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.cep.EventComparator;
@@ -27,6 +28,7 @@
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.nfa.AfterMatchSkipStrategy;
import org.apache.flink.cep.nfa.compiler.NFACompiler;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.util.OutputTag;
@@ -54,8 +56,8 @@ public SelectTimeoutCepOperator(
NFACompiler.NFAFactory nfaFactory,
final EventComparator comparator,
AfterMatchSkipStrategy skipStrategy,
-
[GitHub] dianfu closed pull request #5080: [FLINK-8159] [cep] Add rich support for SelectWrapper and FlatSelectWrapper
dianfu closed pull request #5080: [FLINK-8159] [cep] Add rich support for
SelectWrapper and FlatSelectWrapper
URL: https://github.com/apache/flink/pull/5080
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/FlatSelectTimeoutCepOperator.java
b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/FlatSelectTimeoutCepOperator.java
index 4423bb1dd40..95225271b9e 100644
---
a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/FlatSelectTimeoutCepOperator.java
+++
b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/FlatSelectTimeoutCepOperator.java
@@ -20,7 +20,8 @@
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.api.common.functions.AbstractRichFunction;
+import org.apache.flink.api.common.functions.RichFunction;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.cep.EventComparator;
@@ -28,6 +29,7 @@
import org.apache.flink.cep.PatternFlatTimeoutFunction;
import org.apache.flink.cep.nfa.AfterMatchSkipStrategy;
import org.apache.flink.cep.nfa.compiler.NFACompiler;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.operators.TimestampedCollector;
import org.apache.flink.util.OutputTag;
@@ -102,7 +104,7 @@ protected void processTimedOutSequences(
* in one udf.
*/
@Internal
- public static class FlatSelectWrapper implements
Function {
+ public static class FlatSelectWrapper extends
AbstractRichFunction {
private static final long serialVersionUID =
-8320546120157150202L;
@@ -125,5 +127,25 @@ public FlatSelectWrapper(
this.flatSelectFunction = flatSelectFunction;
this.flatTimeoutFunction = flatTimeoutFunction;
}
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ if (flatSelectFunction instanceof RichFunction) {
+ ((RichFunction)
flatSelectFunction).open(parameters);
+ }
+ if (flatTimeoutFunction instanceof RichFunction) {
+ ((RichFunction)
flatTimeoutFunction).open(parameters);
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (flatSelectFunction instanceof RichFunction) {
+ ((RichFunction) flatSelectFunction).close();
+ }
+ if (flatTimeoutFunction instanceof RichFunction) {
+ ((RichFunction) flatTimeoutFunction).close();
+ }
+ }
}
}
diff --git
a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/SelectTimeoutCepOperator.java
b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/SelectTimeoutCepOperator.java
index cb233a486ec..18a1454cf4c 100644
---
a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/SelectTimeoutCepOperator.java
+++
b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/SelectTimeoutCepOperator.java
@@ -19,7 +19,8 @@
package org.apache.flink.cep.operator;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.api.common.functions.AbstractRichFunction;
+import org.apache.flink.api.common.functions.RichFunction;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.cep.EventComparator;
@@ -27,6 +28,7 @@
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.nfa.AfterMatchSkipStrategy;
import org.apache.flink.cep.nfa.compiler.NFACompiler;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.util.OutputTag;
@@ -54,8 +56,8 @@ public SelectTimeoutCepOperator(
NFACompiler.NFAFactory nfaFactory,
final EventComparator comparator,
AfterMatchSkipStrategy skipStrategy,
- PatternSelectFunction flatSelectFunction,
- PatternTimeoutFunction flatTimeoutFunction,
+ PatternSelectFunction selectFunction,
+ PatternTimeoutFunction timeoutFunction,
OutputTag
[jira] [Commented] (FLINK-10941) Slots prematurely released which still contain unconsumed data
[ https://issues.apache.org/jira/browse/FLINK-10941?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692744#comment-16692744 ] Rafi Aroch commented on FLINK-10941: Saw same issue with Flink 1.5.4. Job fails with same symptoms. > Slots prematurely released which still contain unconsumed data > --- > > Key: FLINK-10941 > URL: https://issues.apache.org/jira/browse/FLINK-10941 > Project: Flink > Issue Type: Bug > Components: ResourceManager >Affects Versions: 1.5.5 >Reporter: Qi >Priority: Major > > Our case is: Flink 1.5 batch mode, 32 parallelism to read data source and 4 > parallelism to write data sink. > > The read task worked perfectly with 32 TMs. However when the job was > executing the write task, since only 4 TMs were needed, other 28 TMs were > released. This caused RemoteTransportException in the write task: > > org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException: > Connection unexpectedly closed by remote task manager > ’the_previous_TM_used_by_read_task'. This might indicate that the remote task > manager was lost. > at > org.apache.flink.runtime.io.network.netty.PartitionRequestClientHandler.channelInactive(PartitionRequestClientHandler.java:133) > at > org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:237) > ... > > After skimming YarnFlinkResourceManager related code, it seems to me that > Flink is releasing TMs when they’re idle, regardless of whether working TMs > need them. > > Put in another way, Flink seems to prematurely release slots which contain > unconsumed data and, thus, eventually release a TM which then fails a > consuming task. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-8159) Pattern(Flat)SelectFunctions should support RichFunction interface
[
https://issues.apache.org/jira/browse/FLINK-8159?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692740#comment-16692740
]
ASF GitHub Bot commented on FLINK-8159:
---
sunjincheng121 commented on a change in pull request #7110: [FLINK-8159] [cep]
Pattern(Flat)SelectFunctions should support RichFunction interface
URL: https://github.com/apache/flink/pull/7110#discussion_r234878521
##
File path:
flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/CepRuntimeContext.java
##
@@ -0,0 +1,196 @@
+/*
+ * 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.cep;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.api.common.accumulators.DoubleCounter;
+import org.apache.flink.api.common.accumulators.Histogram;
+import org.apache.flink.api.common.accumulators.IntCounter;
+import org.apache.flink.api.common.accumulators.LongCounter;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.api.common.functions.BroadcastVariableInitializer;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.state.AggregatingState;
+import org.apache.flink.api.common.state.AggregatingStateDescriptor;
+import org.apache.flink.api.common.state.FoldingState;
+import org.apache.flink.api.common.state.FoldingStateDescriptor;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.state.MapState;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.api.common.state.ReducingState;
+import org.apache.flink.api.common.state.ReducingStateDescriptor;
+import org.apache.flink.api.common.state.ValueState;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.metrics.MetricGroup;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A wrapper class for pattern select function and iterative condition
function's {@link RuntimeContext}.
+ * The runtime context only supports basic operations. Consequently, state
access, accumulators,
+ * broadcast variables and the distributed cache are disabled.
+ */
+public class CepRuntimeContext implements RuntimeContext {
+
+ private final RuntimeContext runtimeContext;
+
+ public CepRuntimeContext(RuntimeContext runtimeContext) {
+ this.runtimeContext = runtimeContext;
+ }
+
+ @Override
+ public String getTaskName() {
+ return runtimeContext.getTaskName();
+ }
+
+ @Override
+ public MetricGroup getMetricGroup() {
+ return runtimeContext.getMetricGroup();
+ }
+
+ @Override
+ public int getNumberOfParallelSubtasks() {
+ return runtimeContext.getNumberOfParallelSubtasks();
+ }
+
+ @Override
+ public int getMaxNumberOfParallelSubtasks() {
+ return runtimeContext.getMaxNumberOfParallelSubtasks();
+ }
+
+ @Override
+ public int getIndexOfThisSubtask() {
+ return runtimeContext.getIndexOfThisSubtask();
+ }
+
+ @Override
+ public int getAttemptNumber() {
+ return runtimeContext.getAttemptNumber();
+ }
+
+ @Override
+ public String getTaskNameWithSubtasks() {
+ return runtimeContext.getTaskNameWithSubtasks();
+ }
+
+ @Override
+ public ExecutionConfig getExecutionConfig() {
+ return runtimeContext.getExecutionConfig();
+ }
+
+ @Override
+ public ClassLoader getUserCodeClassLoader() {
+ return runtimeContext.getUserCodeClassLoader();
+ }
+
+ //
---
+ // Unsupported operations
+ //
---
+
+ @Override
+ public void addAccumulator(
+ String name, Accumulator accumulator)
[jira] [Commented] (FLINK-8159) Pattern(Flat)SelectFunctions should support RichFunction interface
[
https://issues.apache.org/jira/browse/FLINK-8159?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692742#comment-16692742
]
ASF GitHub Bot commented on FLINK-8159:
---
sunjincheng121 commented on a change in pull request #7110: [FLINK-8159] [cep]
Pattern(Flat)SelectFunctions should support RichFunction interface
URL: https://github.com/apache/flink/pull/7110#discussion_r234882983
##
File path:
flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/conditions/RichAndCondition.java
##
@@ -0,0 +1,83 @@
+/*
+ * 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.cep.pattern.conditions;
+
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.functions.util.FunctionUtils;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.Preconditions;
+
+/**
+ * A {@link RichIterativeCondition condition} which combines two conditions
with a logical
+ * {@code AND} and returns {@code true} if both are {@code true}.
+ *
+ * @param Type of the element to filter
+ */
+public class RichAndCondition extends RichIterativeCondition {
Review comment:
I agree to deal with the open and close methods in a way that adds abstract
classes, and RichAnd is very similar to RichOr, but how do we deal with the
difference between RichNot and RichAnd/RichOr?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Pattern(Flat)SelectFunctions should support RichFunction interface
> --
>
> Key: FLINK-8159
> URL: https://issues.apache.org/jira/browse/FLINK-8159
> Project: Flink
> Issue Type: Sub-task
> Components: CEP
>Reporter: Dian Fu
>Assignee: Dian Fu
>Priority: Major
> Labels: pull-request-available
> Fix For: 1.8.0
>
>
> {{SelectWrapper}} and {{FlatSelectWrapper}} should extends
> {{AbstractRichFucntion}} and process properly if the underlying functions
> extend RichFunction.
> Things to be very careful about:
> * backwards compatibility (we previously serialized conditions) - changes to
> those interfaces have to be done carefully
> * we want to be able to add dynamic patterns in the future, so at some point
> we have to open also on control message arrival
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[GitHub] sunjincheng121 commented on a change in pull request #7110: [FLINK-8159] [cep] Pattern(Flat)SelectFunctions should support RichFunction interface
sunjincheng121 commented on a change in pull request #7110: [FLINK-8159] [cep]
Pattern(Flat)SelectFunctions should support RichFunction interface
URL: https://github.com/apache/flink/pull/7110#discussion_r234878521
##
File path:
flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/CepRuntimeContext.java
##
@@ -0,0 +1,196 @@
+/*
+ * 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.cep;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.api.common.accumulators.DoubleCounter;
+import org.apache.flink.api.common.accumulators.Histogram;
+import org.apache.flink.api.common.accumulators.IntCounter;
+import org.apache.flink.api.common.accumulators.LongCounter;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.api.common.functions.BroadcastVariableInitializer;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.state.AggregatingState;
+import org.apache.flink.api.common.state.AggregatingStateDescriptor;
+import org.apache.flink.api.common.state.FoldingState;
+import org.apache.flink.api.common.state.FoldingStateDescriptor;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.state.MapState;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.api.common.state.ReducingState;
+import org.apache.flink.api.common.state.ReducingStateDescriptor;
+import org.apache.flink.api.common.state.ValueState;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.metrics.MetricGroup;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A wrapper class for pattern select function and iterative condition
function's {@link RuntimeContext}.
+ * The runtime context only supports basic operations. Consequently, state
access, accumulators,
+ * broadcast variables and the distributed cache are disabled.
+ */
+public class CepRuntimeContext implements RuntimeContext {
+
+ private final RuntimeContext runtimeContext;
+
+ public CepRuntimeContext(RuntimeContext runtimeContext) {
+ this.runtimeContext = runtimeContext;
+ }
+
+ @Override
+ public String getTaskName() {
+ return runtimeContext.getTaskName();
+ }
+
+ @Override
+ public MetricGroup getMetricGroup() {
+ return runtimeContext.getMetricGroup();
+ }
+
+ @Override
+ public int getNumberOfParallelSubtasks() {
+ return runtimeContext.getNumberOfParallelSubtasks();
+ }
+
+ @Override
+ public int getMaxNumberOfParallelSubtasks() {
+ return runtimeContext.getMaxNumberOfParallelSubtasks();
+ }
+
+ @Override
+ public int getIndexOfThisSubtask() {
+ return runtimeContext.getIndexOfThisSubtask();
+ }
+
+ @Override
+ public int getAttemptNumber() {
+ return runtimeContext.getAttemptNumber();
+ }
+
+ @Override
+ public String getTaskNameWithSubtasks() {
+ return runtimeContext.getTaskNameWithSubtasks();
+ }
+
+ @Override
+ public ExecutionConfig getExecutionConfig() {
+ return runtimeContext.getExecutionConfig();
+ }
+
+ @Override
+ public ClassLoader getUserCodeClassLoader() {
+ return runtimeContext.getUserCodeClassLoader();
+ }
+
+ //
---
+ // Unsupported operations
+ //
---
+
+ @Override
+ public void addAccumulator(
+ String name, Accumulator accumulator) {
+ throw new UnsupportedOperationException("Accumulators are not
supported.");
+ }
+
+ @Override
+ public Accumulator
getAccumulator(String name) {
+ throw new
[GitHub] sunjincheng121 commented on a change in pull request #7110: [FLINK-8159] [cep] Pattern(Flat)SelectFunctions should support RichFunction interface
sunjincheng121 commented on a change in pull request #7110: [FLINK-8159] [cep]
Pattern(Flat)SelectFunctions should support RichFunction interface
URL: https://github.com/apache/flink/pull/7110#discussion_r234882983
##
File path:
flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/conditions/RichAndCondition.java
##
@@ -0,0 +1,83 @@
+/*
+ * 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.cep.pattern.conditions;
+
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.functions.util.FunctionUtils;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.Preconditions;
+
+/**
+ * A {@link RichIterativeCondition condition} which combines two conditions
with a logical
+ * {@code AND} and returns {@code true} if both are {@code true}.
+ *
+ * @param Type of the element to filter
+ */
+public class RichAndCondition extends RichIterativeCondition {
Review comment:
I agree to deal with the open and close methods in a way that adds abstract
classes, and RichAnd is very similar to RichOr, but how do we deal with the
difference between RichNot and RichAnd/RichOr?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[jira] [Commented] (FLINK-8159) Pattern(Flat)SelectFunctions should support RichFunction interface
[
https://issues.apache.org/jira/browse/FLINK-8159?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692741#comment-16692741
]
ASF GitHub Bot commented on FLINK-8159:
---
sunjincheng121 commented on a change in pull request #7110: [FLINK-8159] [cep]
Pattern(Flat)SelectFunctions should support RichFunction interface
URL: https://github.com/apache/flink/pull/7110#discussion_r234885267
##
File path:
flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/RichPatternFlatSelectFunction.java
##
@@ -0,0 +1,57 @@
+/*
+ * 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.cep;
+
+import org.apache.flink.api.common.functions.AbstractRichFunction;
+import org.apache.flink.api.common.functions.RichFunction;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.Preconditions;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Rich variant of the {@link PatternFlatSelectFunction}. As a {@link
RichFunction}, it gives access to the
+ * {@link org.apache.flink.api.common.functions.RuntimeContext} and provides
setup and teardown methods:
+ * {@link RichFunction#open(org.apache.flink.configuration.Configuration)} and
+ * {@link RichFunction#close()}.
+ *
+ * @param Type of the input elements
+ * @param Type of the output element
+ */
+public abstract class RichPatternFlatSelectFunction
Review comment:
Add the test case in `CEPITCase`.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Pattern(Flat)SelectFunctions should support RichFunction interface
> --
>
> Key: FLINK-8159
> URL: https://issues.apache.org/jira/browse/FLINK-8159
> Project: Flink
> Issue Type: Sub-task
> Components: CEP
>Reporter: Dian Fu
>Assignee: Dian Fu
>Priority: Major
> Labels: pull-request-available
> Fix For: 1.8.0
>
>
> {{SelectWrapper}} and {{FlatSelectWrapper}} should extends
> {{AbstractRichFucntion}} and process properly if the underlying functions
> extend RichFunction.
> Things to be very careful about:
> * backwards compatibility (we previously serialized conditions) - changes to
> those interfaces have to be done carefully
> * we want to be able to add dynamic patterns in the future, so at some point
> we have to open also on control message arrival
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-8159) Pattern(Flat)SelectFunctions should support RichFunction interface
[
https://issues.apache.org/jira/browse/FLINK-8159?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692739#comment-16692739
]
ASF GitHub Bot commented on FLINK-8159:
---
sunjincheng121 commented on a change in pull request #7110: [FLINK-8159] [cep]
Pattern(Flat)SelectFunctions should support RichFunction interface
URL: https://github.com/apache/flink/pull/7110#discussion_r234885688
##
File path:
flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/conditions/RichNotCondition.java
##
@@ -0,0 +1,69 @@
+/*
+ * 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.cep.pattern.conditions;
+
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.functions.util.FunctionUtils;
+import org.apache.flink.configuration.Configuration;
+
+/**
+ * A {@link RichIterativeCondition condition} which negates the condition it
wraps
+ * and returns {@code true} if the original condition returns {@code false}.
+ *
+ * @param Type of the element to filter
+ */
+public class RichNotCondition extends RichIterativeCondition {
+
+ private static final long serialVersionUID = 1L;
+
+ private final IterativeCondition original;
+
+ public RichNotCondition(final IterativeCondition original) {
+ this.original = original;
Review comment:
I think we should unify the default behavior for null in `RichOr` ，
`RichAnd` and `RichNot` What do you think? @dianfu @dawidwys
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Pattern(Flat)SelectFunctions should support RichFunction interface
> --
>
> Key: FLINK-8159
> URL: https://issues.apache.org/jira/browse/FLINK-8159
> Project: Flink
> Issue Type: Sub-task
> Components: CEP
>Reporter: Dian Fu
>Assignee: Dian Fu
>Priority: Major
> Labels: pull-request-available
> Fix For: 1.8.0
>
>
> {{SelectWrapper}} and {{FlatSelectWrapper}} should extends
> {{AbstractRichFucntion}} and process properly if the underlying functions
> extend RichFunction.
> Things to be very careful about:
> * backwards compatibility (we previously serialized conditions) - changes to
> those interfaces have to be done carefully
> * we want to be able to add dynamic patterns in the future, so at some point
> we have to open also on control message arrival
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-8159) Pattern(Flat)SelectFunctions should support RichFunction interface
[
https://issues.apache.org/jira/browse/FLINK-8159?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692738#comment-16692738
]
ASF GitHub Bot commented on FLINK-8159:
---
sunjincheng121 commented on a change in pull request #7110: [FLINK-8159] [cep]
Pattern(Flat)SelectFunctions should support RichFunction interface
URL: https://github.com/apache/flink/pull/7110#discussion_r234883989
##
File path:
flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/CepRuntimeContext.java
##
@@ -0,0 +1,196 @@
+/*
+ * 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.cep;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.api.common.accumulators.DoubleCounter;
+import org.apache.flink.api.common.accumulators.Histogram;
+import org.apache.flink.api.common.accumulators.IntCounter;
+import org.apache.flink.api.common.accumulators.LongCounter;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.api.common.functions.BroadcastVariableInitializer;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.state.AggregatingState;
+import org.apache.flink.api.common.state.AggregatingStateDescriptor;
+import org.apache.flink.api.common.state.FoldingState;
+import org.apache.flink.api.common.state.FoldingStateDescriptor;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.state.MapState;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.api.common.state.ReducingState;
+import org.apache.flink.api.common.state.ReducingStateDescriptor;
+import org.apache.flink.api.common.state.ValueState;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.metrics.MetricGroup;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A wrapper class for pattern select function and iterative condition
function's {@link RuntimeContext}.
+ * The runtime context only supports basic operations. Consequently, state
access, accumulators,
+ * broadcast variables and the distributed cache are disabled.
+ */
+public class CepRuntimeContext implements RuntimeContext {
Review comment:
From the points of my view, I like change `CepRuntimeContext` to
`CepFunctionContext`, Because we only need to provide the functions which will
be using in the user-defined function, it will not expose many runtime
functions, such as: `getXXXState` low-level operations. But i am fine if you
want using `CepRuntimeContext` , What do you think @dawidwys @dianfu
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Pattern(Flat)SelectFunctions should support RichFunction interface
> --
>
> Key: FLINK-8159
> URL: https://issues.apache.org/jira/browse/FLINK-8159
> Project: Flink
> Issue Type: Sub-task
> Components: CEP
>Reporter: Dian Fu
>Assignee: Dian Fu
>Priority: Major
> Labels: pull-request-available
> Fix For: 1.8.0
>
>
> {{SelectWrapper}} and {{FlatSelectWrapper}} should extends
> {{AbstractRichFucntion}} and process properly if the underlying functions
> extend RichFunction.
> Things to be very careful about:
> * backwards compatibility (we previously serialized conditions) - changes to
> those interfaces have to be done carefully
> * we want to be able to add dynamic patterns in the future, so at some point
> we have to open also on control message arrival
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[GitHub] sunjincheng121 commented on a change in pull request #7110: [FLINK-8159] [cep] Pattern(Flat)SelectFunctions should support RichFunction interface
sunjincheng121 commented on a change in pull request #7110: [FLINK-8159] [cep]
Pattern(Flat)SelectFunctions should support RichFunction interface
URL: https://github.com/apache/flink/pull/7110#discussion_r234885688
##
File path:
flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/conditions/RichNotCondition.java
##
@@ -0,0 +1,69 @@
+/*
+ * 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.cep.pattern.conditions;
+
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.functions.util.FunctionUtils;
+import org.apache.flink.configuration.Configuration;
+
+/**
+ * A {@link RichIterativeCondition condition} which negates the condition it
wraps
+ * and returns {@code true} if the original condition returns {@code false}.
+ *
+ * @param Type of the element to filter
+ */
+public class RichNotCondition extends RichIterativeCondition {
+
+ private static final long serialVersionUID = 1L;
+
+ private final IterativeCondition original;
+
+ public RichNotCondition(final IterativeCondition original) {
+ this.original = original;
Review comment:
I think we should unify the default behavior for null in `RichOr` ，
`RichAnd` and `RichNot` What do you think? @dianfu @dawidwys
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] sunjincheng121 commented on a change in pull request #7110: [FLINK-8159] [cep] Pattern(Flat)SelectFunctions should support RichFunction interface
sunjincheng121 commented on a change in pull request #7110: [FLINK-8159] [cep]
Pattern(Flat)SelectFunctions should support RichFunction interface
URL: https://github.com/apache/flink/pull/7110#discussion_r234885267
##
File path:
flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/RichPatternFlatSelectFunction.java
##
@@ -0,0 +1,57 @@
+/*
+ * 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.cep;
+
+import org.apache.flink.api.common.functions.AbstractRichFunction;
+import org.apache.flink.api.common.functions.RichFunction;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.Preconditions;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Rich variant of the {@link PatternFlatSelectFunction}. As a {@link
RichFunction}, it gives access to the
+ * {@link org.apache.flink.api.common.functions.RuntimeContext} and provides
setup and teardown methods:
+ * {@link RichFunction#open(org.apache.flink.configuration.Configuration)} and
+ * {@link RichFunction#close()}.
+ *
+ * @param Type of the input elements
+ * @param Type of the output element
+ */
+public abstract class RichPatternFlatSelectFunction
Review comment:
Add the test case in `CEPITCase`.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] sunjincheng121 commented on a change in pull request #7110: [FLINK-8159] [cep] Pattern(Flat)SelectFunctions should support RichFunction interface
sunjincheng121 commented on a change in pull request #7110: [FLINK-8159] [cep]
Pattern(Flat)SelectFunctions should support RichFunction interface
URL: https://github.com/apache/flink/pull/7110#discussion_r234883989
##
File path:
flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/CepRuntimeContext.java
##
@@ -0,0 +1,196 @@
+/*
+ * 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.cep;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.api.common.accumulators.DoubleCounter;
+import org.apache.flink.api.common.accumulators.Histogram;
+import org.apache.flink.api.common.accumulators.IntCounter;
+import org.apache.flink.api.common.accumulators.LongCounter;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.api.common.functions.BroadcastVariableInitializer;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.state.AggregatingState;
+import org.apache.flink.api.common.state.AggregatingStateDescriptor;
+import org.apache.flink.api.common.state.FoldingState;
+import org.apache.flink.api.common.state.FoldingStateDescriptor;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.state.MapState;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.api.common.state.ReducingState;
+import org.apache.flink.api.common.state.ReducingStateDescriptor;
+import org.apache.flink.api.common.state.ValueState;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.metrics.MetricGroup;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A wrapper class for pattern select function and iterative condition
function's {@link RuntimeContext}.
+ * The runtime context only supports basic operations. Consequently, state
access, accumulators,
+ * broadcast variables and the distributed cache are disabled.
+ */
+public class CepRuntimeContext implements RuntimeContext {
Review comment:
From the points of my view, I like change `CepRuntimeContext` to
`CepFunctionContext`, Because we only need to provide the functions which will
be using in the user-defined function, it will not expose many runtime
functions, such as: `getXXXState` low-level operations. But i am fine if you
want using `CepRuntimeContext` , What do you think @dawidwys @dianfu
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[jira] [Commented] (FLINK-7599) Support aggregation functions in the define and measures clause of MatchRecognize
[ https://issues.apache.org/jira/browse/FLINK-7599?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692723#comment-16692723 ] Dian Fu commented on FLINK-7599: Sure, go ahead. > Support aggregation functions in the define and measures clause of > MatchRecognize > - > > Key: FLINK-7599 > URL: https://issues.apache.org/jira/browse/FLINK-7599 > Project: Flink > Issue Type: Sub-task > Components: CEP, Table API SQL >Reporter: Dian Fu >Assignee: Dian Fu >Priority: Major > Fix For: 1.8.0 > > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10892) Flink Yarn app name is hard coded
[ https://issues.apache.org/jira/browse/FLINK-10892?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692715#comment-16692715 ] vinoyang commented on FLINK-10892: -- [~zjffdu] I have already said that Application is just a concept of YARN. In Flink's terminology, there is no such concept, only job. Env.exeute() allows you to set the jobName, and subsequent (not yet merged to the codebase) also allows you to set the jobDescription. If it is not a general concept, I do not recommend providing an API at the env level. In the current mode, if you don't use the API, you can use the CLI option to set it up. I think it is enough to provide this kind of solution. Not everyone uses Flink on YARN. > Flink Yarn app name is hard coded > - > > Key: FLINK-10892 > URL: https://issues.apache.org/jira/browse/FLINK-10892 > Project: Flink > Issue Type: Improvement > Components: YARN >Affects Versions: 1.7.0 >Reporter: Jeff Zhang >Assignee: vinoyang >Priority: Major > > It is always "Flink session cluster", this is not convenient for users to > figure out his corresponding job in Yarn Web UI. Flink should allow user to > specify its yarn app name. > https://github.com/apache/flink/blob/master/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java#L419 -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10868) Flink's Yarn ResourceManager doesn't use yarn.maximum-failed-containers as limit of resource acquirement
[ https://issues.apache.org/jira/browse/FLINK-10868?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692708#comment-16692708 ] Zhenqiu Huang commented on FLINK-10868: --- [~till.rohrmann] I am working on a fix in FlinkYarnResourceManager. In PerJob cluster mode, as mini dispatch will kill itself once the only job stops, it should be easy to stop the cluster by kill the only JobMaster registered in RM with JobMasterGateway. But in session mode, I can only stop each of registered JobMaster when failed containers larger than the threshold set in configuration. Do you have any suggestion to stop session cluster gracefully? > Flink's Yarn ResourceManager doesn't use yarn.maximum-failed-containers as > limit of resource acquirement > > > Key: FLINK-10868 > URL: https://issues.apache.org/jira/browse/FLINK-10868 > Project: Flink > Issue Type: Bug > Components: YARN >Affects Versions: 1.6.2, 1.7.0 >Reporter: Zhenqiu Huang >Assignee: Zhenqiu Huang >Priority: Major > > Currently, YarnResourceManager does use yarn.maximum-failed-containers as > limit of resource acquirement. In worse case, when new start containers > consistently fail, YarnResourceManager will goes into an infinite resource > acquirement process without failing the job. Together with the > https://issues.apache.org/jira/browse/FLINK-10848, It will quick occupy all > resources of yarn queue. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (FLINK-10868) Flink's Yarn ResourceManager doesn't use yarn.maximum-failed-containers as limit of resource acquirement
[ https://issues.apache.org/jira/browse/FLINK-10868?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Zhenqiu Huang updated FLINK-10868: -- Description: Currently, YarnResourceManager does use yarn.maximum-failed-containers as limit of resource acquirement. In worse case, when new start containers consistently fail, YarnResourceManager will goes into an infinite resource acquirement process without failing the job. Together with the https://issues.apache.org/jira/browse/FLINK-10848, It will quick occupy all resources of yarn queue. (was: Currently, YarnResourceManager does use yarn.maximum-failed-containers as limit of resource acquirement. In worse case, when new start containers consistently fail, YarnResourceManager will goes into an infinite resource acquirement process without failing the job. Together with the https://issues.apache.org/jira/browse/FLINK-10848, It will quick occupy all resources of yarn queue. In production, we observe the following that a task manager failed in HA enabled Flink job. At the same time, there is a hdfs failover. During that period, Operation category READ is not supported in state standby. Thus, new acquired task managers keep on failure. ) > Flink's Yarn ResourceManager doesn't use yarn.maximum-failed-containers as > limit of resource acquirement > > > Key: FLINK-10868 > URL: https://issues.apache.org/jira/browse/FLINK-10868 > Project: Flink > Issue Type: Bug > Components: YARN >Affects Versions: 1.6.2, 1.7.0 >Reporter: Zhenqiu Huang >Assignee: Zhenqiu Huang >Priority: Major > > Currently, YarnResourceManager does use yarn.maximum-failed-containers as > limit of resource acquirement. In worse case, when new start containers > consistently fail, YarnResourceManager will goes into an infinite resource > acquirement process without failing the job. Together with the > https://issues.apache.org/jira/browse/FLINK-10848, It will quick occupy all > resources of yarn queue. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-10942) Deduplicate common codes in OutputEmitterTest
zhijiang created FLINK-10942:
Summary: Deduplicate common codes in OutputEmitterTest
Key: FLINK-10942
URL: https://issues.apache.org/jira/browse/FLINK-10942
Project: Flink
Issue Type: Sub-task
Components: Network, Tests
Reporter: zhijiang
Assignee: zhijiang
Fix For: 1.7.1
There are many duplicated codes in {{OutputEmitterTest}} to make it difficult
to maintain. So it is necessary to abstract the common codes to make it simple
which brings benefits for the following refactor work in {{ChannelSelector}}
interface.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-10942) Deduplicate common codes in OutputEmitterTest
[
https://issues.apache.org/jira/browse/FLINK-10942?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692705#comment-16692705
]
ASF GitHub Bot commented on FLINK-10942:
zhijiangW opened a new pull request #7146: [FLINK-10942][network,test]
Deduplicate common codes in OutputEmitterTest
URL: https://github.com/apache/flink/pull/7146
## What is the purpose of the change
*There are many duplicated codes in `OutputEmitterTest` to make it difficult
to maintain. So it is necessary to abstract the common codes to make it simple
which brings benefits for the following refactor work in `ChannelSelector`
interface.*
## Brief change log
- *Abstract common codes for `OutputEmitterTest`*
## Verifying this change
This change is already covered by existing tests, such as
*OutputEmitterTest*.
## 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)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (no)
- If yes, how is the feature documented? (not applicable)
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Deduplicate common codes in OutputEmitterTest
> -
>
> Key: FLINK-10942
> URL: https://issues.apache.org/jira/browse/FLINK-10942
> Project: Flink
> Issue Type: Sub-task
> Components: Network, Tests
>Reporter: zhijiang
>Assignee: zhijiang
>Priority: Minor
> Labels: pull-request-available
> Fix For: 1.7.1
>
>
> There are many duplicated codes in {{OutputEmitterTest}} to make it difficult
> to maintain. So it is necessary to abstract the common codes to make it
> simple which brings benefits for the following refactor work in
> {{ChannelSelector}} interface.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (FLINK-10942) Deduplicate common codes in OutputEmitterTest
[
https://issues.apache.org/jira/browse/FLINK-10942?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-10942:
---
Labels: pull-request-available (was: )
> Deduplicate common codes in OutputEmitterTest
> -
>
> Key: FLINK-10942
> URL: https://issues.apache.org/jira/browse/FLINK-10942
> Project: Flink
> Issue Type: Sub-task
> Components: Network, Tests
>Reporter: zhijiang
>Assignee: zhijiang
>Priority: Minor
> Labels: pull-request-available
> Fix For: 1.7.1
>
>
> There are many duplicated codes in {{OutputEmitterTest}} to make it difficult
> to maintain. So it is necessary to abstract the common codes to make it
> simple which brings benefits for the following refactor work in
> {{ChannelSelector}} interface.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[GitHub] zhijiangW opened a new pull request #7146: [FLINK-10942][network, test] Deduplicate common codes in OutputEmitterTest
zhijiangW opened a new pull request #7146: [FLINK-10942][network,test] Deduplicate common codes in OutputEmitterTest URL: https://github.com/apache/flink/pull/7146 ## What is the purpose of the change *There are many duplicated codes in `OutputEmitterTest` to make it difficult to maintain. So it is necessary to abstract the common codes to make it simple which brings benefits for the following refactor work in `ChannelSelector` interface.* ## Brief change log - *Abstract common codes for `OutputEmitterTest`* ## Verifying this change This change is already covered by existing tests, such as *OutputEmitterTest*. ## 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) - The S3 file system connector: (no) ## Documentation - Does this pull request introduce a new feature? (no) - If yes, how is the feature documented? (not applicable) This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-10892) Flink Yarn app name is hard coded
[ https://issues.apache.org/jira/browse/FLINK-10892?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692701#comment-16692701 ] Jeff Zhang commented on FLINK-10892: We don't need to set it as an yarn app name in api level, but instead via more general setting like this env#setAppName(appName), and internally AbstractYarnClusterDescriptor use this to set yarn app name. > Flink Yarn app name is hard coded > - > > Key: FLINK-10892 > URL: https://issues.apache.org/jira/browse/FLINK-10892 > Project: Flink > Issue Type: Improvement > Components: YARN >Affects Versions: 1.7.0 >Reporter: Jeff Zhang >Assignee: vinoyang >Priority: Major > > It is always "Flink session cluster", this is not convenient for users to > figure out his corresponding job in Yarn Web UI. Flink should allow user to > specify its yarn app name. > https://github.com/apache/flink/blob/master/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java#L419 -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10461) Speed up download file procedure when restore
[ https://issues.apache.org/jira/browse/FLINK-10461?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692689#comment-16692689 ] ASF GitHub Bot commented on FLINK-10461: klion26 commented on issue #6777: [FLINK-10461] [State Backends, Checkpointing] Speed up download files when restore from DFS URL: https://github.com/apache/flink/pull/6777#issuecomment-440143611 Hi, @azagrebin , please help to review this when you have time. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Speed up download file procedure when restore > -- > > Key: FLINK-10461 > URL: https://issues.apache.org/jira/browse/FLINK-10461 > Project: Flink > Issue Type: Improvement > Components: State Backends, Checkpointing >Reporter: Congxian Qiu >Assignee: Congxian Qiu >Priority: Major > Labels: pull-request-available > > In the current master branch, the restore will download file from DFS, the > download procedure are single-thread, this could speed up by using > multi-thread for downloading states from DFS. > > In my company, the states will come to some terabytes, so the restore > procedure will become a litter slow, after a bit digging, I find download > states from DFS using single thread, this could using multi-thread for speed > up. > I test the time used for download states from DFS with ~2 terabytes states. > With single thread it used 640+s, and 130+s when using 5 threads for download. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[GitHub] klion26 commented on issue #6777: [FLINK-10461] [State Backends, Checkpointing] Speed up download files when restore from DFS
klion26 commented on issue #6777: [FLINK-10461] [State Backends, Checkpointing] Speed up download files when restore from DFS URL: https://github.com/apache/flink/pull/6777#issuecomment-440143611 Hi, @azagrebin , please help to review this when you have time. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Created] (FLINK-10941) Slots prematurely released which still contain unconsumed data
Qi created FLINK-10941: -- Summary: Slots prematurely released which still contain unconsumed data Key: FLINK-10941 URL: https://issues.apache.org/jira/browse/FLINK-10941 Project: Flink Issue Type: Bug Components: ResourceManager Affects Versions: 1.5.5 Reporter: Qi Our case is: Flink 1.5 batch mode, 32 parallelism to read data source and 4 parallelism to write data sink. The read task worked perfectly with 32 TMs. However when the job was executing the write task, since only 4 TMs were needed, other 28 TMs were released. This caused RemoteTransportException in the write task: org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException: Connection unexpectedly closed by remote task manager ’the_previous_TM_used_by_read_task'. This might indicate that the remote task manager was lost. at org.apache.flink.runtime.io.network.netty.PartitionRequestClientHandler.channelInactive(PartitionRequestClientHandler.java:133) at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:237) ... After skimming YarnFlinkResourceManager related code, it seems to me that Flink is releasing TMs when they’re idle, regardless of whether working TMs need them. Put in another way, Flink seems to prematurely release slots which contain unconsumed data and, thus, eventually release a TM which then fails a consuming task. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (FLINK-10922) Refactor the placement of the Flink Kafka connector end to end test module
[ https://issues.apache.org/jira/browse/FLINK-10922?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] vinoyang updated FLINK-10922: - Fix Version/s: 1.7.0 > Refactor the placement of the Flink Kafka connector end to end test module > -- > > Key: FLINK-10922 > URL: https://issues.apache.org/jira/browse/FLINK-10922 > Project: Flink > Issue Type: Sub-task > Components: E2E Tests >Reporter: vinoyang >Assignee: vinoyang >Priority: Critical > Labels: pull-request-available > Fix For: 1.7.0 > > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10940) Refactor the maven dependencies of Kafka 0.8 connector
[
https://issues.apache.org/jira/browse/FLINK-10940?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692670#comment-16692670
]
ASF GitHub Bot commented on FLINK-10940:
yanghua closed pull request #7145: [FLINK-10940] Refactor the maven
dependencies of Kafka 0.8 connector
URL: https://github.com/apache/flink/pull/7145
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/flink-connectors/flink-connector-kafka-0.8/pom.xml
b/flink-connectors/flink-connector-kafka-0.8/pom.xml
index b3a65931869..e16321310c5 100644
--- a/flink-connectors/flink-connector-kafka-0.8/pom.xml
+++ b/flink-connectors/flink-connector-kafka-0.8/pom.xml
@@ -63,8 +63,10 @@ under the License.
flink-connector-kafka-base_${scala.binary.version}
${project.version}
-
+
+ org.apache.kafka
+
kafka_${scala.binary.version}
+
org.apache.kafka
kafka-clients
@@ -99,46 +101,8 @@ under the License.
org.apache.kafka
- kafka_${scala.binary.version}
+ kafka-clients
${kafka.version}
-
-
- com.sun.jmx
- jmxri
-
-
- com.sun.jdmk
- jmxtools
-
-
- log4j
- log4j
-
-
- org.slf4j
- slf4j-simple
-
-
- net.sf.jopt-simple
- jopt-simple
-
-
- org.scala-lang
- scala-reflect
-
-
- org.scala-lang
- scala-compiler
-
-
- com.yammer.metrics
-
metrics-annotation
-
-
- org.xerial.snappy
- snappy-java
-
-
@@ -187,6 +151,20 @@ under the License.
${project.version}
test-jar
test
+
+
+
+ org.apache.kafka
+
kafka_${scala.binary.version}
+
+
+
+
+
+ org.apache.kafka
+ kafka_${scala.binary.version}
+ ${kafka.version}
+ test
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Refactor the maven dependencies of Kafka 0.8 connector
> --
>
> Key: FLINK-10940
> URL: https://issues.apache.org/jira/browse/FLINK-10940
> Project: Flink
> Issue Type: Improvement
> Components: Kafka Connector
>Reporter: vinoyang
>Assignee: vinoyang
>Priority: Major
> Labels: pull-request-available
>
> Currently, Kafka 0.9, 0.10, 0.11, the modern connector's dependent form
> remains uniform, but Kafka 0.8 is a special case. This issue will refactor
> it, and make it consistent with
[jira] [Closed] (FLINK-10940) Refactor the maven dependencies of Kafka 0.8 connector
[ https://issues.apache.org/jira/browse/FLINK-10940?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] vinoyang closed FLINK-10940. Resolution: Won't Fix Release Note: In-depth, Kafka 0.8 connector is indeed different from other versions, this refactoring cannot be completed. > Refactor the maven dependencies of Kafka 0.8 connector > -- > > Key: FLINK-10940 > URL: https://issues.apache.org/jira/browse/FLINK-10940 > Project: Flink > Issue Type: Improvement > Components: Kafka Connector >Reporter: vinoyang >Assignee: vinoyang >Priority: Major > Labels: pull-request-available > > Currently, Kafka 0.9, 0.10, 0.11, the modern connector's dependent form > remains uniform, but Kafka 0.8 is a special case. This issue will refactor > it, and make it consistent with other versions of the connector. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10892) Flink Yarn app name is hard coded
[ https://issues.apache.org/jira/browse/FLINK-10892?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692669#comment-16692669 ] vinoyang commented on FLINK-10892: -- [~zjffdu] Flink on YARN is just one of the running models supported by Flink, although this scenario is perhaps the most popular. So this API is not suitable for inclusion in the ExecutionEnvironment. > Flink Yarn app name is hard coded > - > > Key: FLINK-10892 > URL: https://issues.apache.org/jira/browse/FLINK-10892 > Project: Flink > Issue Type: Improvement > Components: YARN >Affects Versions: 1.7.0 >Reporter: Jeff Zhang >Assignee: vinoyang >Priority: Major > > It is always "Flink session cluster", this is not convenient for users to > figure out his corresponding job in Yarn Web UI. Flink should allow user to > specify its yarn app name. > https://github.com/apache/flink/blob/master/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java#L419 -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Comment Edited] (FLINK-10928) Job unable to stabilise after restart
[
https://issues.apache.org/jira/browse/FLINK-10928?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692618#comment-16692618
]
Biao Liu edited comment on FLINK-10928 at 11/20/18 4:08 AM:
Hi [~djharper]
1. "Why does YARN kill the containers with out of memory?"
The reason is described clearly in exception.
{code:java}
Container [pid=7725,containerID=container_1541433014652_0001_01_
000716] is running beyond physical memory limits. Current usage: 6.4 GB of 6.4
GB physical memory used; 8.4 GB of 31.9 GB virtual memory used. Killing
container.{code}
Your container is beyond physical memory limits. Not because OOM, OOM may
cause failure, but not being killed.
2. "Is it possible for the task manager to allocate memory outside of the 'off
heap' allocation, which would cause YARN to kill the container?"
Yes, it is possible. JVM, state backend, Netty, all these components may
allocate off heap memory or native memory.
3. "Why do we get timeout waiting for connection from pool from the AWS SDK?"
I'm not sure because I can't see the whole picture of your job. However there
is a "FileNotFoundException" which is thrown by user code. I think that's not
caused by Flink, right?
{code:java}
Caused by: org.apache.beam.sdk.util.UserCodeException:
java.io.FileNotFoundException: Reopen at position 0 on
s3a://.../beam/.temp-beam-2018-11-05_15-54-26-0/bc47b14b-1679-45ce-81b7-
a4d19e036cb5: com.amazonaws.services.s3.model.AmazonS3Exception: The specified
key does not exist. (Service: Amazon S3; Status Code: 404; Error Code:
NoSuchKey; Request ID: 0D67ACD1037E5B52; S3 Extended Request ID:
BVgqzksS75Dv1EkZyUgkVMl8brE1PznBM1RsN9uXp2cnn8Rf+r+b9D09TWZQtpW8aSbQi7R9 RW8=),
S3 Extended Request ID:
BVgqzksS75Dv1EkZyUgkVMl8brE1PznBM1RsN9uXp2cnn8Rf+r+b9D09TWZQtpW8aSbQi7R9 RW8=
{code}
There are too many problems in your description. Most of them seem to be
nothing related with Flink framework.
Could you fix the memory and the FileNotFoundException first?
And also I think this should be answered in Flink user mailing list not here.
was (Author: sleepy):
Hi [~djharper]
1. "Why does YARN kill the containers with out of memory?"
The reason is described clearly in exception.
Container [pid=7725,containerID=container_1541433014652_0001_01_
000716] is running beyond physical memory limits. Current usage: 6.4 GB of 6.4
GB physical memory used; 8.4 GB of 31.9 GB virtual memory used. Killing
container.
Your container is beyond physical memory limits. Not because OOM, OOM may cause
failure, but not being killed.
2. "Is it possible for the task manager to allocate memory outside of the 'off
heap' allocation, which would cause YARN to kill the container?"
Yes, it is possible. JVM, state backend, Netty, all these components may
allocate off heap memory or native memory.
3. "Why do we get timeout waiting for connection from pool from the AWS SDK?"
I'm not sure because I can't see the whole picture of your job. However there
is a "FileNotFoundException" which is thrown by user code. I think that's not
caused by Flink, right?
{code:java}
Caused by: org.apache.beam.sdk.util.UserCodeException:
java.io.FileNotFoundException: Reopen at position 0 on
s3a://.../beam/.temp-beam-2018-11-05_15-54-26-0/bc47b14b-1679-45ce-81b7-
a4d19e036cb5: com.amazonaws.services.s3.model.AmazonS3Exception: The specified
key does not exist. (Service: Amazon S3; Status Code: 404; Error Code:
NoSuchKey; Request ID: 0D67ACD1037E5B52; S3 Extended Request ID:
BVgqzksS75Dv1EkZyUgkVMl8brE1PznBM1RsN9uXp2cnn8Rf+r+b9D09TWZQtpW8aSbQi7R9 RW8=),
S3 Extended Request ID:
BVgqzksS75Dv1EkZyUgkVMl8brE1PznBM1RsN9uXp2cnn8Rf+r+b9D09TWZQtpW8aSbQi7R9 RW8=
{code}
There are too many problems in your description. Most of them seem to be
nothing related with Flink framework.
Could you fix the memory and the FileNotFoundException first?
> Job unable to stabilise after restart
> --
>
> Key: FLINK-10928
> URL: https://issues.apache.org/jira/browse/FLINK-10928
> Project: Flink
> Issue Type: Bug
> Environment: AWS EMR 5.17.0
> FLINK 1.5.2
> BEAM 2.7.0
>Reporter: Daniel Harper
>Priority: Major
> Attachments: Screen Shot 2018-11-16 at 15.49.03.png, Screen Shot
> 2018-11-16 at 15.49.15.png,
> ants-CopyofThe'death'spiralincident-191118-1231-1332.pdf
>
>
> We've seen a few instances of this occurring in production now (it's
> difficult to reproduce)
> I've attached a timeline of events as a PDF here
> [^ants-CopyofThe'death'spiralincident-191118-1231-1332.pdf] but essentially
> it boils down to
> 1. Job restarts due to exception
> 2. Job restores from a checkpoint but we get the exception
> {code}
> Caused by: com.amazonaws.SdkClientException: Unable to execute HTTP request:
> Timeout waiting for
[jira] [Commented] (FLINK-10928) Job unable to stabilise after restart
[
https://issues.apache.org/jira/browse/FLINK-10928?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692618#comment-16692618
]
Biao Liu commented on FLINK-10928:
--
Hi [~djharper]
1. "Why does YARN kill the containers with out of memory?"
The reason is described clearly in exception.
Container [pid=7725,containerID=container_1541433014652_0001_01_
000716] is running beyond physical memory limits. Current usage: 6.4 GB of 6.4
GB physical memory used; 8.4 GB of 31.9 GB virtual memory used. Killing
container.
Your container is beyond physical memory limits. Not because OOM, OOM may cause
failure, but not being killed.
2. "Is it possible for the task manager to allocate memory outside of the 'off
heap' allocation, which would cause YARN to kill the container?"
Yes, it is possible. JVM, state backend, Netty, all these components may
allocate off heap memory or native memory.
3. "Why do we get timeout waiting for connection from pool from the AWS SDK?"
I'm not sure because I can't see the whole picture of your job. However there
is a "FileNotFoundException" which is thrown by user code. I think that's not
caused by Flink, right?
{code:java}
Caused by: org.apache.beam.sdk.util.UserCodeException:
java.io.FileNotFoundException: Reopen at position 0 on
s3a://.../beam/.temp-beam-2018-11-05_15-54-26-0/bc47b14b-1679-45ce-81b7-
a4d19e036cb5: com.amazonaws.services.s3.model.AmazonS3Exception: The specified
key does not exist. (Service: Amazon S3; Status Code: 404; Error Code:
NoSuchKey; Request ID: 0D67ACD1037E5B52; S3 Extended Request ID:
BVgqzksS75Dv1EkZyUgkVMl8brE1PznBM1RsN9uXp2cnn8Rf+r+b9D09TWZQtpW8aSbQi7R9 RW8=),
S3 Extended Request ID:
BVgqzksS75Dv1EkZyUgkVMl8brE1PznBM1RsN9uXp2cnn8Rf+r+b9D09TWZQtpW8aSbQi7R9 RW8=
{code}
There are too many problems in your description. Most of them seem to be
nothing related with Flink framework.
Could you fix the memory and the FileNotFoundException first?
> Job unable to stabilise after restart
> --
>
> Key: FLINK-10928
> URL: https://issues.apache.org/jira/browse/FLINK-10928
> Project: Flink
> Issue Type: Bug
> Environment: AWS EMR 5.17.0
> FLINK 1.5.2
> BEAM 2.7.0
>Reporter: Daniel Harper
>Priority: Major
> Attachments: Screen Shot 2018-11-16 at 15.49.03.png, Screen Shot
> 2018-11-16 at 15.49.15.png,
> ants-CopyofThe'death'spiralincident-191118-1231-1332.pdf
>
>
> We've seen a few instances of this occurring in production now (it's
> difficult to reproduce)
> I've attached a timeline of events as a PDF here
> [^ants-CopyofThe'death'spiralincident-191118-1231-1332.pdf] but essentially
> it boils down to
> 1. Job restarts due to exception
> 2. Job restores from a checkpoint but we get the exception
> {code}
> Caused by: com.amazonaws.SdkClientException: Unable to execute HTTP request:
> Timeout waiting for connection from pool
> {code}
> 3. Job restarts
> 4. Job restores from a checkpoint but we get the same exception
> repeat a few times within 2-3 minutes
> 5. YARN kills containers with out of memory
> {code}
> 2018-11-14 00:16:04,430 INFO org.apache.flink.yarn.YarnResourceManager
> - Closing TaskExecutor connection
> container_1541433014652_0001_01_000716 because: Container
> [pid=7725,containerID=container_1541433014652_0001_01_
> 000716] is running beyond physical memory limits. Current usage: 6.4 GB of
> 6.4 GB physical memory used; 8.4 GB of 31.9 GB virtual memory used. Killing
> container.
> Dump of the process-tree for container_1541433014652_0001_01_000716 :
> |- PID PPID PGRPID SESSID CMD_NAME USER_MODE_TIME(MILLIS)
> SYSTEM_TIME(MILLIS) VMEM_USAGE(BYTES) RSSMEM_USAGE(PAGES) FULL_CMD_LINE
> |- 7725 7723 7725 7725 (bash) 0 0 115863552 696 /bin/bash -c
> /usr/lib/jvm/java-openjdk/bin/java -Xms4995m -Xmx4995m
> -XX:MaxDirectMemorySize=1533m
> -Xloggc:/var/log/hadoop-yarn/flink_gc_container_1541433014652_0001_%p.log
> -XX:GCLogF
> ileSize=200M -XX:NumberOfGCLogFiles=10 -XX:+PrintGCDetails
> -XX:+PrintGCTimeStamps -XX:+PrintTenuringDistribution -XX:+PrintGCCause
> -XX:+PrintGCDateStamps -XX:+UseG1GC
> -Dlog.file=/var/log/hadoop-yarn/containers/application_1541433014652_00
> 01/container_1541433014652_0001_01_000716/taskmanager.log
> -Dlog4j.configuration=file:./log4j.properties
> org.apache.flink.yarn.YarnTaskExecutorRunner --configDir . 1>
> /var/log/hadoop-yarn/containers/application_1541433014652_0001/container
> _1541433014652_0001_01_000716/taskmanager.out 2>
> /var/log/hadoop-yarn/containers/application_1541433014652_0001/container_1541433014652_0001_01_000716/taskmanager.err
> |- 7738 7725 7725 7725 (java) 6959576 976377 8904458240 1671684
> /usr/lib/jvm/java-openjdk/bin/java -Xms4995m -Xmx4995m
> -XX:MaxDirectMemorySize=1533m
>
[jira] [Comment Edited] (FLINK-9043) Introduce a friendly way to resume the job from externalized checkpoints automatically
[
https://issues.apache.org/jira/browse/FLINK-9043?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692610#comment-16692610
]
Yun Tang edited comment on FLINK-9043 at 11/20/18 3:55 AM:
---
I quite agree with [~sihuazhou]'s idea, user just need to provide the
checkpoint directory at job-id level.
Currently, user must know the latest checkpoint id to resume job with command
below:
{code:bash}
$ bin/flink run -s {/user-definied-path/job-id/chk-id}
{code}
We could simplify this with given previous job-id, so that the job would resume
from latest checkpoint automatically.
{code:bash}
$ bin/flink run -r {/user-definied-path/job-id}
{code}
was (Author: yunta):
I quite agreed with [~sihuazhou]'s idea, user just need to provide the
checkpoint directory at job-id level.
Currently, user must know the latest checkpoint id to resume job with command
below:
{code:bash}
$ bin/flink run -s {/user-definied-path/job-id/chk-id}
{code}
We could simplify this with given previous job-id, so that the job would resume
from latest checkpoint automatically.
{code:bash}
$ bin/flink run -r {/user-definied-path/job-id}
{code}
> Introduce a friendly way to resume the job from externalized checkpoints
> automatically
> --
>
> Key: FLINK-9043
> URL: https://issues.apache.org/jira/browse/FLINK-9043
> Project: Flink
> Issue Type: New Feature
>Reporter: godfrey johnson
>Assignee: Sihua Zhou
>Priority: Major
> Labels: pull-request-available
>
> I know a flink job can reovery from checkpoint with restart strategy, but can
> not recovery as spark streaming jobs when job is starting.
> Every time, the submitted flink job is regarded as a new job, while , in the
> spark streaming job, which can detect the checkpoint directory first, and
> then recovery from the latest succeed one. However, Flink only can recovery
> until the job failed first, then retry with strategy.
>
> So, would flink support to recover from the checkpoint directly in a new job?
> h2. New description by [~sihuazhou]
> Currently, it's quite a bit not friendly for users to recover job from the
> externalized checkpoint, user need to find the dedicate dir for the job which
> is not a easy thing when there are too many jobs. This ticket attend to
> introduce a more friendly way to allow the user to use the externalized
> checkpoint to do recovery.
> The implementation steps are copied from the comments of [~StephanEwen]:
> - We could make this an option where you pass a flag (-r) to automatically
> look for the latest checkpoint in a given directory.
> - If more than one jobs checkpointed there before, this operation would fail.
> - We might also need a way to have jobs not create the UUID subdirectory,
> otherwise the scanning for the latest checkpoint would not easily work.
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9043) Introduce a friendly way to resume the job from externalized checkpoints automatically
[
https://issues.apache.org/jira/browse/FLINK-9043?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692610#comment-16692610
]
Yun Tang commented on FLINK-9043:
-
I quite agreed with [~sihuazhou]'s idea, user just need to provide the
checkpoint directory at job-id level.
Currently, user must know the latest checkpoint id to resume job with command
below:
{code:bash}
$ bin/flink run -s {/user-definied-path/job-id/chk-id}
{code}
We could simplify this with given previous job-id, so that the job would resume
from latest checkpoint automatically.
{code:bash}
$ bin/flink run -r {/user-definied-path/job-id}
{code}
> Introduce a friendly way to resume the job from externalized checkpoints
> automatically
> --
>
> Key: FLINK-9043
> URL: https://issues.apache.org/jira/browse/FLINK-9043
> Project: Flink
> Issue Type: New Feature
>Reporter: godfrey johnson
>Assignee: Sihua Zhou
>Priority: Major
> Labels: pull-request-available
>
> I know a flink job can reovery from checkpoint with restart strategy, but can
> not recovery as spark streaming jobs when job is starting.
> Every time, the submitted flink job is regarded as a new job, while , in the
> spark streaming job, which can detect the checkpoint directory first, and
> then recovery from the latest succeed one. However, Flink only can recovery
> until the job failed first, then retry with strategy.
>
> So, would flink support to recover from the checkpoint directly in a new job?
> h2. New description by [~sihuazhou]
> Currently, it's quite a bit not friendly for users to recover job from the
> externalized checkpoint, user need to find the dedicate dir for the job which
> is not a easy thing when there are too many jobs. This ticket attend to
> introduce a more friendly way to allow the user to use the externalized
> checkpoint to do recovery.
> The implementation steps are copied from the comments of [~StephanEwen]:
> - We could make this an option where you pass a flag (-r) to automatically
> look for the latest checkpoint in a given directory.
> - If more than one jobs checkpointed there before, this operation would fail.
> - We might also need a way to have jobs not create the UUID subdirectory,
> otherwise the scanning for the latest checkpoint would not easily work.
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-10892) Flink Yarn app name is hard coded
[ https://issues.apache.org/jira/browse/FLINK-10892?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692609#comment-16692609 ] Jeff Zhang commented on FLINK-10892: [~yanghua] Shouldn't we expose it in ExecutionEnvironment, AbstractYarnClusterDescriptor is a low level api, very few people know this I believe. > Flink Yarn app name is hard coded > - > > Key: FLINK-10892 > URL: https://issues.apache.org/jira/browse/FLINK-10892 > Project: Flink > Issue Type: Improvement > Components: YARN >Affects Versions: 1.7.0 >Reporter: Jeff Zhang >Assignee: vinoyang >Priority: Major > > It is always "Flink session cluster", this is not convenient for users to > figure out his corresponding job in Yarn Web UI. Flink should allow user to > specify its yarn app name. > https://github.com/apache/flink/blob/master/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java#L419 -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Closed] (FLINK-10892) Flink Yarn app name is hard coded
[ https://issues.apache.org/jira/browse/FLINK-10892?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] vinoyang closed FLINK-10892. Resolution: Not A Problem Release Note: We can set it via CLI option or API (AbstractYarnClusterDescriptor#setName). > Flink Yarn app name is hard coded > - > > Key: FLINK-10892 > URL: https://issues.apache.org/jira/browse/FLINK-10892 > Project: Flink > Issue Type: Improvement > Components: YARN >Affects Versions: 1.7.0 >Reporter: Jeff Zhang >Assignee: vinoyang >Priority: Major > > It is always "Flink session cluster", this is not convenient for users to > figure out his corresponding job in Yarn Web UI. Flink should allow user to > specify its yarn app name. > https://github.com/apache/flink/blob/master/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java#L419 -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10892) Flink Yarn app name is hard coded
[ https://issues.apache.org/jira/browse/FLINK-10892?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692607#comment-16692607 ] vinoyang commented on FLINK-10892: -- [~dawidwys] is right. He gave an option based on the config option. If using API, you can call [setName() |https://github.com/apache/flink/blob/master/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java#L1325] if you given a custom name, it would be used, see [here|https://github.com/apache/flink/blob/master/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java#L1030]. > Flink Yarn app name is hard coded > - > > Key: FLINK-10892 > URL: https://issues.apache.org/jira/browse/FLINK-10892 > Project: Flink > Issue Type: Improvement > Components: YARN >Affects Versions: 1.7.0 >Reporter: Jeff Zhang >Assignee: vinoyang >Priority: Major > > It is always "Flink session cluster", this is not convenient for users to > figure out his corresponding job in Yarn Web UI. Flink should allow user to > specify its yarn app name. > https://github.com/apache/flink/blob/master/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java#L419 -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10940) Refactor the maven dependencies of Kafka 0.8 connector
[ https://issues.apache.org/jira/browse/FLINK-10940?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692598#comment-16692598 ] ASF GitHub Bot commented on FLINK-10940: yanghua opened a new pull request #7145: [FLINK-10940] Refactor the maven dependencies of Kafka 0.8 connector URL: https://github.com/apache/flink/pull/7145 ## What is the purpose of the change *This pull request refactors the maven dependencies of Kafka 0.8 connector* ## Brief change log - *Refactor the maven dependencies of Kafka 0.8 connector* ## 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): (**yes** / no) - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**) - The serializers: (yes / **no** / don't know) - The runtime per-record code paths (performance sensitive): (yes / **no** / don't know) - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes / **no** / don't know) - The S3 file system connector: (yes / **no** / don't know) ## Documentation - Does this pull request introduce a new feature? (yes / **no**) - If yes, how is the feature documented? (not applicable / docs / JavaDocs / **not documented**) This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Refactor the maven dependencies of Kafka 0.8 connector > -- > > Key: FLINK-10940 > URL: https://issues.apache.org/jira/browse/FLINK-10940 > Project: Flink > Issue Type: Improvement > Components: Kafka Connector >Reporter: vinoyang >Assignee: vinoyang >Priority: Major > Labels: pull-request-available > > Currently, Kafka 0.9, 0.10, 0.11, the modern connector's dependent form > remains uniform, but Kafka 0.8 is a special case. This issue will refactor > it, and make it consistent with other versions of the connector. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (FLINK-10940) Refactor the maven dependencies of Kafka 0.8 connector
[ https://issues.apache.org/jira/browse/FLINK-10940?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] ASF GitHub Bot updated FLINK-10940: --- Labels: pull-request-available (was: ) > Refactor the maven dependencies of Kafka 0.8 connector > -- > > Key: FLINK-10940 > URL: https://issues.apache.org/jira/browse/FLINK-10940 > Project: Flink > Issue Type: Improvement > Components: Kafka Connector >Reporter: vinoyang >Assignee: vinoyang >Priority: Major > Labels: pull-request-available > > Currently, Kafka 0.9, 0.10, 0.11, the modern connector's dependent form > remains uniform, but Kafka 0.8 is a special case. This issue will refactor > it, and make it consistent with other versions of the connector. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[GitHub] yanghua opened a new pull request #7145: [FLINK-10940] Refactor the maven dependencies of Kafka 0.8 connector
yanghua opened a new pull request #7145: [FLINK-10940] Refactor the maven dependencies of Kafka 0.8 connector URL: https://github.com/apache/flink/pull/7145 ## What is the purpose of the change *This pull request refactors the maven dependencies of Kafka 0.8 connector* ## Brief change log - *Refactor the maven dependencies of Kafka 0.8 connector* ## 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): (**yes** / no) - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**) - The serializers: (yes / **no** / don't know) - The runtime per-record code paths (performance sensitive): (yes / **no** / don't know) - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes / **no** / don't know) - The S3 file system connector: (yes / **no** / don't know) ## Documentation - Does this pull request introduce a new feature? (yes / **no**) - If yes, how is the feature documented? (not applicable / docs / JavaDocs / **not documented**) This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Created] (FLINK-10940) Refactor the maven dependencies of Kafka 0.8 connector
vinoyang created FLINK-10940: Summary: Refactor the maven dependencies of Kafka 0.8 connector Key: FLINK-10940 URL: https://issues.apache.org/jira/browse/FLINK-10940 Project: Flink Issue Type: Improvement Components: Kafka Connector Reporter: vinoyang Assignee: vinoyang Currently, Kafka 0.9, 0.10, 0.11, the modern connector's dependent form remains uniform, but Kafka 0.8 is a special case. This issue will refactor it, and make it consistent with other versions of the connector. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10926) Fix the problem for function TIMESTAMPDIFF in Table
[
https://issues.apache.org/jira/browse/FLINK-10926?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692592#comment-16692592
]
xuqianjin commented on FLINK-10926:
---
[~xueyu7452]
Well, I will check again when the version of 1.7 comes out.
Thank you very much
> Fix the problem for function TIMESTAMPDIFF in Table
> ---
>
> Key: FLINK-10926
> URL: https://issues.apache.org/jira/browse/FLINK-10926
> Project: Flink
> Issue Type: Bug
> Components: Table API SQL
>Affects Versions: 1.6.2
>Reporter: xuqianjin
>Priority: Minor
> Attachments: image-2018-11-19-18-33-47-389.png,
> image-2018-11-19-22-23-09-554.png
>
>
> Use the following SQL statement:
> val result3 = tEnv.sqlQuery("select TIMESTAMPDIFF(MINUTE,'2012-08-24
> 09:00:00','2012-08-30 12:00:00')")
> The following errors occurred:
> Caused by: org.apache.calcite.runtime.CalciteContextException: From line 1,
> column 8 to line 1, column 72: No match found for function signature
> TIMESTAMPDIFF(, , )
> at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
> Expect to be able to return the time difference correctly
>
> val result3 = tEnv.sqlQuery("select TIMESTAMPDIFF (MINUTE, TIMESTAMP
> '2012-08-24 09:00:00', TIMESTAMP '2012-08-30 12:00:00')")
> Caused by: org.apache.calcite.runtime.CalciteContextException: From line 1,
> column 8 to line 1, column 95: No match found for function signature
> TIMESTAMPDIFF(, , )
> at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-10926) Fix the problem for function TIMESTAMPDIFF in Table
[
https://issues.apache.org/jira/browse/FLINK-10926?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692586#comment-16692586
]
xueyu commented on FLINK-10926:
---
hi, [~x1q1j1], Version 1.6.2 does not support timestampdiff. This function is
added recently about Sep 26, maybe in upcoming release 1.7.0. This PR
(https://github.com/apache/flink/pull/6282) added TimestampDiff in Sql and
Table API. So you could rebase on master or wait 1.7.0 which will be released
these days. Thanks~
> Fix the problem for function TIMESTAMPDIFF in Table
> ---
>
> Key: FLINK-10926
> URL: https://issues.apache.org/jira/browse/FLINK-10926
> Project: Flink
> Issue Type: Bug
> Components: Table API SQL
>Affects Versions: 1.6.2
>Reporter: xuqianjin
>Priority: Minor
> Attachments: image-2018-11-19-18-33-47-389.png,
> image-2018-11-19-22-23-09-554.png
>
>
> Use the following SQL statement:
> val result3 = tEnv.sqlQuery("select TIMESTAMPDIFF(MINUTE,'2012-08-24
> 09:00:00','2012-08-30 12:00:00')")
> The following errors occurred:
> Caused by: org.apache.calcite.runtime.CalciteContextException: From line 1,
> column 8 to line 1, column 72: No match found for function signature
> TIMESTAMPDIFF(, , )
> at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
> Expect to be able to return the time difference correctly
>
> val result3 = tEnv.sqlQuery("select TIMESTAMPDIFF (MINUTE, TIMESTAMP
> '2012-08-24 09:00:00', TIMESTAMP '2012-08-30 12:00:00')")
> Caused by: org.apache.calcite.runtime.CalciteContextException: From line 1,
> column 8 to line 1, column 95: No match found for function signature
> TIMESTAMPDIFF(, , )
> at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Assigned] (FLINK-10929) Add support for Apache Arrow
[ https://issues.apache.org/jira/browse/FLINK-10929?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] vinoyang reassigned FLINK-10929: Assignee: vinoyang > Add support for Apache Arrow > > > Key: FLINK-10929 > URL: https://issues.apache.org/jira/browse/FLINK-10929 > Project: Flink > Issue Type: Wish >Reporter: Pedro Cardoso Silva >Assignee: vinoyang >Priority: Minor > > Investigate the possibility of adding support for Apache Arrow as a > standardized columnar, memory format for data. > Given the activity that [https://github.com/apache/arrow] is currently > getting and its claims objective of providing a zero-copy, standardized data > format across platforms, I think it makes sense for Flink to look into > supporting it. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10926) Fix the problem for function TIMESTAMPDIFF in Table
[
https://issues.apache.org/jira/browse/FLINK-10926?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692571#comment-16692571
]
xuqianjin commented on FLINK-10926:
---
hi [~xueyu7452] It was found in version 1.6.2 but I'm sure it's the same in
other versions
> Fix the problem for function TIMESTAMPDIFF in Table
> ---
>
> Key: FLINK-10926
> URL: https://issues.apache.org/jira/browse/FLINK-10926
> Project: Flink
> Issue Type: Bug
> Components: Table API SQL
>Affects Versions: 1.6.2
>Reporter: xuqianjin
>Priority: Minor
> Attachments: image-2018-11-19-18-33-47-389.png,
> image-2018-11-19-22-23-09-554.png
>
>
> Use the following SQL statement:
> val result3 = tEnv.sqlQuery("select TIMESTAMPDIFF(MINUTE,'2012-08-24
> 09:00:00','2012-08-30 12:00:00')")
> The following errors occurred:
> Caused by: org.apache.calcite.runtime.CalciteContextException: From line 1,
> column 8 to line 1, column 72: No match found for function signature
> TIMESTAMPDIFF(, , )
> at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
> Expect to be able to return the time difference correctly
>
> val result3 = tEnv.sqlQuery("select TIMESTAMPDIFF (MINUTE, TIMESTAMP
> '2012-08-24 09:00:00', TIMESTAMP '2012-08-30 12:00:00')")
> Caused by: org.apache.calcite.runtime.CalciteContextException: From line 1,
> column 8 to line 1, column 95: No match found for function signature
> TIMESTAMPDIFF(, , )
> at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (FLINK-10926) Fix the problem for function TIMESTAMPDIFF in Table
[
https://issues.apache.org/jira/browse/FLINK-10926?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
xuqianjin updated FLINK-10926:
--
Affects Version/s: 1.6.2
> Fix the problem for function TIMESTAMPDIFF in Table
> ---
>
> Key: FLINK-10926
> URL: https://issues.apache.org/jira/browse/FLINK-10926
> Project: Flink
> Issue Type: Bug
> Components: Table API SQL
>Affects Versions: 1.6.2
>Reporter: xuqianjin
>Priority: Minor
> Attachments: image-2018-11-19-18-33-47-389.png,
> image-2018-11-19-22-23-09-554.png
>
>
> Use the following SQL statement:
> val result3 = tEnv.sqlQuery("select TIMESTAMPDIFF(MINUTE,'2012-08-24
> 09:00:00','2012-08-30 12:00:00')")
> The following errors occurred:
> Caused by: org.apache.calcite.runtime.CalciteContextException: From line 1,
> column 8 to line 1, column 72: No match found for function signature
> TIMESTAMPDIFF(, , )
> at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
> Expect to be able to return the time difference correctly
>
> val result3 = tEnv.sqlQuery("select TIMESTAMPDIFF (MINUTE, TIMESTAMP
> '2012-08-24 09:00:00', TIMESTAMP '2012-08-30 12:00:00')")
> Caused by: org.apache.calcite.runtime.CalciteContextException: From line 1,
> column 8 to line 1, column 95: No match found for function signature
> TIMESTAMPDIFF(, , )
> at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-10919) Flink binary distribution too large
[
https://issues.apache.org/jira/browse/FLINK-10919?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692569#comment-16692569
]
vinoyang commented on FLINK-10919:
--
[~till.rohrmann] Currently, the typical pattern of Flink combined with Python
is Apache Beam (Python SDK) + Flink Runner. I suggest that we can remove the
Python-related sample jars from the example module. If you agree with this
idea, we can create an issue to track it.
> Flink binary distribution too large
> ---
>
> Key: FLINK-10919
> URL: https://issues.apache.org/jira/browse/FLINK-10919
> Project: Flink
> Issue Type: Improvement
> Components: Build System
>Affects Versions: 1.7.0
>Reporter: Till Rohrmann
>Priority: Major
> Fix For: 1.8.0
>
>
> Flink's binary distribution is almost 500 MB large:
> {code}
> 128K ./bin
> 16K ./examples/python/streaming
> 28K ./examples/python/batch
> 44K ./examples/python
> 189M ./examples/streaming
> 240K ./examples/gelly
> 136K ./examples/batch
> 190M ./examples
> 131M ./lib
> 174M ./opt
> 0B ./log
> 56K ./conf
> 494M .
> {code}
> I think this is far too large and we should try to reduce the size. For
> example, the examples directory contains 3 different Kafka example jobs, each
> at least 50 MB of size.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-10932) Initial flink-kubernetes module with empty implementation
[
https://issues.apache.org/jira/browse/FLINK-10932?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692542#comment-16692542
]
ASF GitHub Bot commented on FLINK-10932:
TisonKun commented on a change in pull request #7144: [FLINK-10932]
[ResourceManager] Initial flink-kubernetes module with empty implementation
URL: https://github.com/apache/flink/pull/7144#discussion_r234843453
##
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/cluster/KubernetesClusterDescriptor.java
##
@@ -0,0 +1,151 @@
+/*
+ * 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.kubernetes.cluster;
+
+import org.apache.flink.client.deployment.ClusterDeploymentException;
+import org.apache.flink.client.deployment.ClusterDescriptor;
+import org.apache.flink.client.deployment.ClusterRetrieveException;
+import org.apache.flink.client.deployment.ClusterSpecification;
+import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.client.program.rest.RestClusterClient;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.kubernetes.FlinkKubernetesOptions;
+import org.apache.flink.kubernetes.kubeclient.Endpoint;
+import org.apache.flink.kubernetes.kubeclient.KubeClient;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.util.FlinkException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import javax.annotation.Nonnull;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * Kubernetes specific {@link ClusterDescriptor} implementation.
+ */
+public class KubernetesClusterDescriptor implements ClusterDescriptor {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(KubernetesClusterDescriptor.class);
+
+ private static final String CLUSTER_ID_PREFIX =
"flink-session-cluster-";
+
+ private static final String CLUSTER_DESCRIPTION = "Kubernetes cluster";
+
+ private FlinkKubernetesOptions options;
+
+ private KubeClient client;
+
+ public KubernetesClusterDescriptor(@Nonnull FlinkKubernetesOptions
options, @Nonnull KubeClient client) {
+ this.options = options;
+ this.client = client;
+ }
+
+ private String generateClusterId(){
Review comment:
`) {`
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Initial flink-kubernetes module with empty implementation
> -
>
> Key: FLINK-10932
> URL: https://issues.apache.org/jira/browse/FLINK-10932
> Project: Flink
> Issue Type: Sub-task
> Components: ResourceManager
>Reporter: JIN SUN
>Assignee: JIN SUN
>Priority: Major
> Labels: pull-request-available
>
> Initial the skeleton module to start native kubernetes integration.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Assigned] (FLINK-9461) Disentangle flink-connector-kafka from flink-table and flink-json
[
https://issues.apache.org/jira/browse/FLINK-9461?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
vinoyang reassigned FLINK-9461:
---
Assignee: vinoyang
> Disentangle flink-connector-kafka from flink-table and flink-json
> -
>
> Key: FLINK-9461
> URL: https://issues.apache.org/jira/browse/FLINK-9461
> Project: Flink
> Issue Type: Improvement
> Components: Build System
>Affects Versions: 1.5.0
>Reporter: Till Rohrmann
>Assignee: vinoyang
>Priority: Major
> Fix For: 1.8.0
>
>
> Currently, the {{flink-connector-kafka}} module has a dependency on
> {{flink-table}} and {{flink-json}}. The reason seems to be that the module
> contains the {{KafkaJsonTableSource}} and {{KafkaJsonTableSink}}. Even though
> the {{flink-table}} and {{flink-json}} dependency are marked as optional, the
> {{flink-connector-kafka}} will still contain the table sources and sinks. I
> think this is not a clean design.
> I would propose to move the table sources and sinks into a dedicated module
> which depends on {{flink-connector-kafka}}. That way we would better separate
> dependencies and could remove {{flink-table}} and {{flink-json}} from
> {{flink-connector-kafka}}.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-10932) Initial flink-kubernetes module with empty implementation
[
https://issues.apache.org/jira/browse/FLINK-10932?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692541#comment-16692541
]
ASF GitHub Bot commented on FLINK-10932:
TisonKun commented on a change in pull request #7144: [FLINK-10932]
[ResourceManager] Initial flink-kubernetes module with empty implementation
URL: https://github.com/apache/flink/pull/7144#discussion_r234843511
##
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/cluster/KubernetesClusterDescriptor.java
##
@@ -0,0 +1,151 @@
+/*
+ * 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.kubernetes.cluster;
+
+import org.apache.flink.client.deployment.ClusterDeploymentException;
+import org.apache.flink.client.deployment.ClusterDescriptor;
+import org.apache.flink.client.deployment.ClusterRetrieveException;
+import org.apache.flink.client.deployment.ClusterSpecification;
+import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.client.program.rest.RestClusterClient;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.kubernetes.FlinkKubernetesOptions;
+import org.apache.flink.kubernetes.kubeclient.Endpoint;
+import org.apache.flink.kubernetes.kubeclient.KubeClient;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.util.FlinkException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import javax.annotation.Nonnull;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * Kubernetes specific {@link ClusterDescriptor} implementation.
+ */
+public class KubernetesClusterDescriptor implements ClusterDescriptor {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(KubernetesClusterDescriptor.class);
+
+ private static final String CLUSTER_ID_PREFIX =
"flink-session-cluster-";
+
+ private static final String CLUSTER_DESCRIPTION = "Kubernetes cluster";
+
+ private FlinkKubernetesOptions options;
+
+ private KubeClient client;
+
+ public KubernetesClusterDescriptor(@Nonnull FlinkKubernetesOptions
options, @Nonnull KubeClient client) {
+ this.options = options;
+ this.client = client;
+ }
+
+ private String generateClusterId(){
+ return CLUSTER_ID_PREFIX + UUID.randomUUID();
+ }
+
+ @Override
+ public String getClusterDescription() {
+ return CLUSTER_DESCRIPTION;
+ }
+
+ private ClusterClient createClusterEndpoint(Endpoint
clusterEndpoint, String clusterId) throws Exception {
+
+ Configuration configuration = new
Configuration(this.options.getConfiguration());
+ configuration.setString(JobManagerOptions.ADDRESS,
clusterEndpoint.getAddress());
+ configuration.setInteger(JobManagerOptions.PORT,
clusterEndpoint.getPort());
+ return new RestClusterClient<>(configuration, clusterId);
+ }
+
+ @Override
+ public ClusterClient retrieve(String clusterId) throws
ClusterRetrieveException {
+ try {
+ Endpoint clusterEndpoint =
this.client.getResetEndpoint(clusterId);
+ return this.createClusterEndpoint(clusterEndpoint,
clusterId);
+ } catch (Exception e) {
+ throw new ClusterRetrieveException("Could not create
the RestClusterClient.", e);
+ }
+ }
+
+ @Override
+ public ClusterClient deploySessionCluster(ClusterSpecification
clusterSpecification)
+ throws ClusterDeploymentException {
+
+ String clusterId = this.generateClusterId();
+
+ //TODO: add arguments
+ final List args = Arrays.asList();
+
+ return this.deployClusterInternal(clusterId, args);
+ }
+
+ @Override
+ public ClusterClient deployJobCluster(ClusterSpecification
clusterSpecification, JobGraph jobGraph, boolean detached) {
+
[jira] [Commented] (FLINK-10932) Initial flink-kubernetes module with empty implementation
[
https://issues.apache.org/jira/browse/FLINK-10932?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692540#comment-16692540
]
ASF GitHub Bot commented on FLINK-10932:
TisonKun commented on a change in pull request #7144: [FLINK-10932]
[ResourceManager] Initial flink-kubernetes module with empty implementation
URL: https://github.com/apache/flink/pull/7144#discussion_r234843542
##
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/cluster/KubernetesClusterDescriptor.java
##
@@ -0,0 +1,151 @@
+/*
+ * 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.kubernetes.cluster;
+
+import org.apache.flink.client.deployment.ClusterDeploymentException;
+import org.apache.flink.client.deployment.ClusterDescriptor;
+import org.apache.flink.client.deployment.ClusterRetrieveException;
+import org.apache.flink.client.deployment.ClusterSpecification;
+import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.client.program.rest.RestClusterClient;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.kubernetes.FlinkKubernetesOptions;
+import org.apache.flink.kubernetes.kubeclient.Endpoint;
+import org.apache.flink.kubernetes.kubeclient.KubeClient;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.util.FlinkException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import javax.annotation.Nonnull;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * Kubernetes specific {@link ClusterDescriptor} implementation.
+ */
+public class KubernetesClusterDescriptor implements ClusterDescriptor {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(KubernetesClusterDescriptor.class);
+
+ private static final String CLUSTER_ID_PREFIX =
"flink-session-cluster-";
+
+ private static final String CLUSTER_DESCRIPTION = "Kubernetes cluster";
+
+ private FlinkKubernetesOptions options;
+
+ private KubeClient client;
+
+ public KubernetesClusterDescriptor(@Nonnull FlinkKubernetesOptions
options, @Nonnull KubeClient client) {
+ this.options = options;
+ this.client = client;
+ }
+
+ private String generateClusterId(){
+ return CLUSTER_ID_PREFIX + UUID.randomUUID();
+ }
+
+ @Override
+ public String getClusterDescription() {
+ return CLUSTER_DESCRIPTION;
+ }
+
+ private ClusterClient createClusterEndpoint(Endpoint
clusterEndpoint, String clusterId) throws Exception {
+
+ Configuration configuration = new
Configuration(this.options.getConfiguration());
+ configuration.setString(JobManagerOptions.ADDRESS,
clusterEndpoint.getAddress());
+ configuration.setInteger(JobManagerOptions.PORT,
clusterEndpoint.getPort());
+ return new RestClusterClient<>(configuration, clusterId);
+ }
+
+ @Override
+ public ClusterClient retrieve(String clusterId) throws
ClusterRetrieveException {
+ try {
+ Endpoint clusterEndpoint =
this.client.getResetEndpoint(clusterId);
+ return this.createClusterEndpoint(clusterEndpoint,
clusterId);
+ } catch (Exception e) {
+ throw new ClusterRetrieveException("Could not create
the RestClusterClient.", e);
+ }
+ }
+
+ @Override
+ public ClusterClient deploySessionCluster(ClusterSpecification
clusterSpecification)
+ throws ClusterDeploymentException {
+
+ String clusterId = this.generateClusterId();
+
+ //TODO: add arguments
+ final List args = Arrays.asList();
+
+ return this.deployClusterInternal(clusterId, args);
+ }
+
+ @Override
+ public ClusterClient deployJobCluster(ClusterSpecification
clusterSpecification, JobGraph jobGraph, boolean detached) {
+
[GitHub] TisonKun commented on a change in pull request #7144: [FLINK-10932] [ResourceManager] Initial flink-kubernetes module with empty implementation
TisonKun commented on a change in pull request #7144: [FLINK-10932]
[ResourceManager] Initial flink-kubernetes module with empty implementation
URL: https://github.com/apache/flink/pull/7144#discussion_r234843453
##
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/cluster/KubernetesClusterDescriptor.java
##
@@ -0,0 +1,151 @@
+/*
+ * 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.kubernetes.cluster;
+
+import org.apache.flink.client.deployment.ClusterDeploymentException;
+import org.apache.flink.client.deployment.ClusterDescriptor;
+import org.apache.flink.client.deployment.ClusterRetrieveException;
+import org.apache.flink.client.deployment.ClusterSpecification;
+import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.client.program.rest.RestClusterClient;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.kubernetes.FlinkKubernetesOptions;
+import org.apache.flink.kubernetes.kubeclient.Endpoint;
+import org.apache.flink.kubernetes.kubeclient.KubeClient;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.util.FlinkException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import javax.annotation.Nonnull;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * Kubernetes specific {@link ClusterDescriptor} implementation.
+ */
+public class KubernetesClusterDescriptor implements ClusterDescriptor {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(KubernetesClusterDescriptor.class);
+
+ private static final String CLUSTER_ID_PREFIX =
"flink-session-cluster-";
+
+ private static final String CLUSTER_DESCRIPTION = "Kubernetes cluster";
+
+ private FlinkKubernetesOptions options;
+
+ private KubeClient client;
+
+ public KubernetesClusterDescriptor(@Nonnull FlinkKubernetesOptions
options, @Nonnull KubeClient client) {
+ this.options = options;
+ this.client = client;
+ }
+
+ private String generateClusterId(){
Review comment:
`) {`
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] TisonKun commented on a change in pull request #7144: [FLINK-10932] [ResourceManager] Initial flink-kubernetes module with empty implementation
TisonKun commented on a change in pull request #7144: [FLINK-10932]
[ResourceManager] Initial flink-kubernetes module with empty implementation
URL: https://github.com/apache/flink/pull/7144#discussion_r234843542
##
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/cluster/KubernetesClusterDescriptor.java
##
@@ -0,0 +1,151 @@
+/*
+ * 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.kubernetes.cluster;
+
+import org.apache.flink.client.deployment.ClusterDeploymentException;
+import org.apache.flink.client.deployment.ClusterDescriptor;
+import org.apache.flink.client.deployment.ClusterRetrieveException;
+import org.apache.flink.client.deployment.ClusterSpecification;
+import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.client.program.rest.RestClusterClient;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.kubernetes.FlinkKubernetesOptions;
+import org.apache.flink.kubernetes.kubeclient.Endpoint;
+import org.apache.flink.kubernetes.kubeclient.KubeClient;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.util.FlinkException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import javax.annotation.Nonnull;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * Kubernetes specific {@link ClusterDescriptor} implementation.
+ */
+public class KubernetesClusterDescriptor implements ClusterDescriptor {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(KubernetesClusterDescriptor.class);
+
+ private static final String CLUSTER_ID_PREFIX =
"flink-session-cluster-";
+
+ private static final String CLUSTER_DESCRIPTION = "Kubernetes cluster";
+
+ private FlinkKubernetesOptions options;
+
+ private KubeClient client;
+
+ public KubernetesClusterDescriptor(@Nonnull FlinkKubernetesOptions
options, @Nonnull KubeClient client) {
+ this.options = options;
+ this.client = client;
+ }
+
+ private String generateClusterId(){
+ return CLUSTER_ID_PREFIX + UUID.randomUUID();
+ }
+
+ @Override
+ public String getClusterDescription() {
+ return CLUSTER_DESCRIPTION;
+ }
+
+ private ClusterClient createClusterEndpoint(Endpoint
clusterEndpoint, String clusterId) throws Exception {
+
+ Configuration configuration = new
Configuration(this.options.getConfiguration());
+ configuration.setString(JobManagerOptions.ADDRESS,
clusterEndpoint.getAddress());
+ configuration.setInteger(JobManagerOptions.PORT,
clusterEndpoint.getPort());
+ return new RestClusterClient<>(configuration, clusterId);
+ }
+
+ @Override
+ public ClusterClient retrieve(String clusterId) throws
ClusterRetrieveException {
+ try {
+ Endpoint clusterEndpoint =
this.client.getResetEndpoint(clusterId);
+ return this.createClusterEndpoint(clusterEndpoint,
clusterId);
+ } catch (Exception e) {
+ throw new ClusterRetrieveException("Could not create
the RestClusterClient.", e);
+ }
+ }
+
+ @Override
+ public ClusterClient deploySessionCluster(ClusterSpecification
clusterSpecification)
+ throws ClusterDeploymentException {
+
+ String clusterId = this.generateClusterId();
+
+ //TODO: add arguments
+ final List args = Arrays.asList();
+
+ return this.deployClusterInternal(clusterId, args);
+ }
+
+ @Override
+ public ClusterClient deployJobCluster(ClusterSpecification
clusterSpecification, JobGraph jobGraph, boolean detached) {
+ throw new NotImplementedException();
+ }
+
+ @Nonnull
+ private ClusterClient deployClusterInternal(String clusterId,
List args) throws ClusterDeploymentException {
+ try {
+
[GitHub] TisonKun commented on a change in pull request #7144: [FLINK-10932] [ResourceManager] Initial flink-kubernetes module with empty implementation
TisonKun commented on a change in pull request #7144: [FLINK-10932]
[ResourceManager] Initial flink-kubernetes module with empty implementation
URL: https://github.com/apache/flink/pull/7144#discussion_r234843511
##
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/cluster/KubernetesClusterDescriptor.java
##
@@ -0,0 +1,151 @@
+/*
+ * 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.kubernetes.cluster;
+
+import org.apache.flink.client.deployment.ClusterDeploymentException;
+import org.apache.flink.client.deployment.ClusterDescriptor;
+import org.apache.flink.client.deployment.ClusterRetrieveException;
+import org.apache.flink.client.deployment.ClusterSpecification;
+import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.client.program.rest.RestClusterClient;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.kubernetes.FlinkKubernetesOptions;
+import org.apache.flink.kubernetes.kubeclient.Endpoint;
+import org.apache.flink.kubernetes.kubeclient.KubeClient;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.util.FlinkException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import javax.annotation.Nonnull;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * Kubernetes specific {@link ClusterDescriptor} implementation.
+ */
+public class KubernetesClusterDescriptor implements ClusterDescriptor {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(KubernetesClusterDescriptor.class);
+
+ private static final String CLUSTER_ID_PREFIX =
"flink-session-cluster-";
+
+ private static final String CLUSTER_DESCRIPTION = "Kubernetes cluster";
+
+ private FlinkKubernetesOptions options;
+
+ private KubeClient client;
+
+ public KubernetesClusterDescriptor(@Nonnull FlinkKubernetesOptions
options, @Nonnull KubeClient client) {
+ this.options = options;
+ this.client = client;
+ }
+
+ private String generateClusterId(){
+ return CLUSTER_ID_PREFIX + UUID.randomUUID();
+ }
+
+ @Override
+ public String getClusterDescription() {
+ return CLUSTER_DESCRIPTION;
+ }
+
+ private ClusterClient createClusterEndpoint(Endpoint
clusterEndpoint, String clusterId) throws Exception {
+
+ Configuration configuration = new
Configuration(this.options.getConfiguration());
+ configuration.setString(JobManagerOptions.ADDRESS,
clusterEndpoint.getAddress());
+ configuration.setInteger(JobManagerOptions.PORT,
clusterEndpoint.getPort());
+ return new RestClusterClient<>(configuration, clusterId);
+ }
+
+ @Override
+ public ClusterClient retrieve(String clusterId) throws
ClusterRetrieveException {
+ try {
+ Endpoint clusterEndpoint =
this.client.getResetEndpoint(clusterId);
+ return this.createClusterEndpoint(clusterEndpoint,
clusterId);
+ } catch (Exception e) {
+ throw new ClusterRetrieveException("Could not create
the RestClusterClient.", e);
+ }
+ }
+
+ @Override
+ public ClusterClient deploySessionCluster(ClusterSpecification
clusterSpecification)
+ throws ClusterDeploymentException {
+
+ String clusterId = this.generateClusterId();
+
+ //TODO: add arguments
+ final List args = Arrays.asList();
+
+ return this.deployClusterInternal(clusterId, args);
+ }
+
+ @Override
+ public ClusterClient deployJobCluster(ClusterSpecification
clusterSpecification, JobGraph jobGraph, boolean detached) {
+ throw new NotImplementedException();
+ }
+
+ @Nonnull
+ private ClusterClient deployClusterInternal(String clusterId,
List args) throws ClusterDeploymentException {
+ try {
+
[jira] [Commented] (FLINK-10931) Exclude transitive dependence on Kafka-client
[
https://issues.apache.org/jira/browse/FLINK-10931?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692438#comment-16692438
]
ASF GitHub Bot commented on FLINK-10931:
yanghua commented on issue #7142: [FLINK-10931] Exclude transitive dependence
on Kafka-client
URL: https://github.com/apache/flink/pull/7142#issuecomment-440084229
@zentol It's strange. Yesterday, in my own Travis, the CI for FLINK-10922
was failed. The maven reported these error message :
```
12:42:38.301 [WARNING] Rule 0:
org.apache.maven.plugins.enforcer.DependencyConvergence failed with message:
Failed while enforcing releasability. See above detailed error message.
```
and
```
12:42:38.494 [ERROR] Failed to execute goal
org.apache.maven.plugins:maven-enforcer-plugin:3.0.0-M1:enforce
(dependency-convergence) on project flink-streaming-kafka010-test_2.11: Some
Enforcer rules have failed. Look above for specific messages explaining why the
rule failed. -> [Help 1]
```
It seems @tillrohrmann has fixed this issue and marked it as the part of
FLINK-10922.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Exclude transitive dependence on Kafka-client
> -
>
> Key: FLINK-10931
> URL: https://issues.apache.org/jira/browse/FLINK-10931
> Project: Flink
> Issue Type: Improvement
> Components: Kafka Connector
>Reporter: vinoyang
>Assignee: vinoyang
>Priority: Major
> Labels: pull-request-available
>
> log detail : [https://api.travis-ci.org/v3/job/456937363/log.txt]
>
> {code:java}
> Dependency convergence error for org.apache.kafka:kafka-clients:0.10.2.1
> paths to dependency are:
> +-org.apache.flink:flink-streaming-kafka010-test_2.11:1.8-SNAPSHOT
> +-org.apache.flink:flink-connector-kafka-0.10_2.11:1.8-SNAPSHOT
> +-org.apache.flink:flink-connector-kafka-0.9_2.11:1.8-SNAPSHOT
> +-org.apache.flink:flink-connector-kafka-base_2.11:1.8-SNAPSHOT
> +-org.apache.kafka:kafka-clients:0.10.2.1
> and
> +-org.apache.flink:flink-streaming-kafka010-test_2.11:1.8-SNAPSHOT
> +-org.apache.flink:flink-connector-kafka-0.10_2.11:1.8-SNAPSHOT
> +-org.apache.flink:flink-connector-kafka-0.9_2.11:1.8-SNAPSHOT
> +-org.apache.kafka:kafka-clients:0.9.0.1
> and
> +-org.apache.flink:flink-streaming-kafka010-test_2.11:1.8-SNAPSHOT
> +-org.apache.flink:flink-connector-kafka-0.10_2.11:1.8-SNAPSHOT
> +-org.apache.kafka:kafka-clients:0.10.2.1
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[GitHub] yanghua commented on issue #7142: [FLINK-10931] Exclude transitive dependence on Kafka-client
yanghua commented on issue #7142: [FLINK-10931] Exclude transitive dependence on Kafka-client URL: https://github.com/apache/flink/pull/7142#issuecomment-440084229 @zentol It's strange. Yesterday, in my own Travis, the CI for FLINK-10922 was failed. The maven reported these error message : ``` 12:42:38.301 [WARNING] Rule 0: org.apache.maven.plugins.enforcer.DependencyConvergence failed with message: Failed while enforcing releasability. See above detailed error message. ``` and ``` 12:42:38.494 [ERROR] Failed to execute goal org.apache.maven.plugins:maven-enforcer-plugin:3.0.0-M1:enforce (dependency-convergence) on project flink-streaming-kafka010-test_2.11: Some Enforcer rules have failed. Look above for specific messages explaining why the rule failed. -> [Help 1] ``` It seems @tillrohrmann has fixed this issue and marked it as the part of FLINK-10922. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Comment Edited] (FLINK-9541) Add robots.txt and sitemap.xml to Flink website
[
https://issues.apache.org/jira/browse/FLINK-9541?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16689936#comment-16689936
]
Ken Krugler edited comment on FLINK-9541 at 11/19/18 11:38 PM:
---
I'd asked [on the bui...@apache.org|mailto:on%c2%a0the%c2%a0bui...@apache.org]
about setting this up, but didn't hear back. Turns out Gavin McDonald had
[responded|http://mail-archives.apache.org/mod_mbox/www-builds/201806.mbox/%3C21B85DEA-438A-42F0-8FAE-F25820F396A9%4016degrees.com.au%3E]...
{quote}Ok Ken and anyone else interested. I have updated the robots.txt [1]
file to point to a sitemap-index.xml [2] file. So, all you now need to do is
ensure you have a flink.xml.gz sitemap in ci.apache.org/projects/flink
<[http://ci.apache.org/projects/flink]> and create a PR against our
sitemap-index.xml file, and done, hopefully.
{quote}
I can create the sitemap file and build the pull request, but it would be good
to get some input on what to put in the sitemap. For example, as a first cut it
would be easiest to just have
[https://ci.apache.org/projects/flink/flink-docs-stable/] as the only docs, as
(I assume) that's what we'd want most people to find if they were doing a
search without a version number in the query, yes? Maybe [~fhueske] can weigh
in here.
was (Author: kkrugler):
I'd asked [on the bui...@apache.org|mailto:on%c2%a0the%c2%a0bui...@apache.org]
about setting this up, but didn't hear back. Turns out Gavin McDonald had
responded..[.|http://mail-archives.apache.org/mod_mbox/www-builds/201806.mbox/%3C21B85DEA-438A-42F0-8FAE-F25820F396A9%4016degrees.com.au%3E]
{quote}Ok Ken and anyone else interested. I have updated the robots.txt [1]
file to point to a sitemap-index.xml [2] file. So, all you now need to do is
ensure you have a flink.xml.gz sitemap in ci.apache.org/projects/flink
<[http://ci.apache.org/projects/flink]> and create a PR against our
sitemap-index.xml file, and done, hopefully.
{quote}
I can create the sitemap file and build the pull request, but it would be good
to get some input on what to put in the sitemap. For example, as a first cut it
would be easiest to just have
[https://ci.apache.org/projects/flink/flink-docs-stable/] as the only docs, as
(I assume) that's what we'd want most people to find if they were doing a
search without a version number in the query, yes? Maybe [~fhueske] can weigh
in here.
> Add robots.txt and sitemap.xml to Flink website
> ---
>
> Key: FLINK-9541
> URL: https://issues.apache.org/jira/browse/FLINK-9541
> Project: Flink
> Issue Type: Improvement
> Components: Project Website
>Reporter: Fabian Hueske
>Priority: Major
>
> From the [dev mailing
> list|https://lists.apache.org/thread.html/71ce1bfbed1cf5f0069b27a46df1cd4dccbe8abefa75ac85601b088b@%3Cdev.flink.apache.org%3E]:
> {quote}
> It would help to add a sitemap (and the robots.txt required to reference it)
> for flink.apache.org and ci.apache.org (for /projects/flink)
> You can see what Tomcat did along these lines -
> http://tomcat.apache.org/robots.txt references
> http://tomcat.apache.org/sitemap.xml, which is a sitemap index file pointing
> to http://tomcat.apache.org/sitemap-main.xml
> By doing this, you can emphasize more recent versions of docs. There are
> other benefits, but reducing poor Google search results (to me) is the
> biggest win.
> E.g. https://www.google.com/search?q=flink+reducingstate
> (search on flink
> reducing state) shows the 1.3 Javadocs (hit #1), master (1.6-SNAPSHOT)
> Javadocs (hit #2), and then many pages of other results.
> Whereas the Javadocs for 1.5
>
> and 1.4
>
> are nowhere to be found.
> {quote}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Comment Edited] (FLINK-9541) Add robots.txt and sitemap.xml to Flink website
[
https://issues.apache.org/jira/browse/FLINK-9541?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16689936#comment-16689936
]
Ken Krugler edited comment on FLINK-9541 at 11/19/18 11:37 PM:
---
I'd asked [on the bui...@apache.org|mailto:on%c2%a0the%c2%a0bui...@apache.org]
about setting this up, but didn't hear back. Turns out Gavin McDonald had
responded..[.|http://mail-archives.apache.org/mod_mbox/www-builds/201806.mbox/%3C21B85DEA-438A-42F0-8FAE-F25820F396A9%4016degrees.com.au%3E]
{quote}Ok Ken and anyone else interested. I have updated the robots.txt [1]
file to point to a sitemap-index.xml [2] file. So, all you now need to do is
ensure you have a flink.xml.gz sitemap in ci.apache.org/projects/flink
<[http://ci.apache.org/projects/flink]> and create a PR against our
sitemap-index.xml file, and done, hopefully.
{quote}
I can create the sitemap file and build the pull request, but it would be good
to get some input on what to put in the sitemap. For example, as a first cut it
would be easiest to just have
[https://ci.apache.org/projects/flink/flink-docs-stable/] as the only docs, as
(I assume) that's what we'd want most people to find if they were doing a
search without a version number in the query, yes? Maybe [~fhueske] can weigh
in here.
was (Author: kkrugler):
I'd asked [on the bui...@apache.org|mailto:on%c2%a0the%c2%a0bui...@apache.org]
about setting this up, but didn't hear back. Turns out Gavin McDonald had
responded...
{quote}Ok Ken and anyone else interested. I have updated the robots.txt [1]
file to point to a sitemap-index.xml [2] file. So, all you now need to do is
ensure you have a flink.xml.gz sitemap in ci.apache.org/projects/flink
and create a PR against our
sitemap-index.xml file, and done, hopefully.
{quote}
I can create the sitemap file and build the pull request, but it would be good
to get some input on what to put in the sitemap. For example, as a first cut it
would be easiest to just have
[https://ci.apache.org/projects/flink/flink-docs-stable/] as the only docs, as
(I assume) that's what we'd want most people to find if they were doing a
search without a version number in the query, yes? Maybe [~fhueske] can weigh
in here.
> Add robots.txt and sitemap.xml to Flink website
> ---
>
> Key: FLINK-9541
> URL: https://issues.apache.org/jira/browse/FLINK-9541
> Project: Flink
> Issue Type: Improvement
> Components: Project Website
>Reporter: Fabian Hueske
>Priority: Major
>
> From the [dev mailing
> list|https://lists.apache.org/thread.html/71ce1bfbed1cf5f0069b27a46df1cd4dccbe8abefa75ac85601b088b@%3Cdev.flink.apache.org%3E]:
> {quote}
> It would help to add a sitemap (and the robots.txt required to reference it)
> for flink.apache.org and ci.apache.org (for /projects/flink)
> You can see what Tomcat did along these lines -
> http://tomcat.apache.org/robots.txt references
> http://tomcat.apache.org/sitemap.xml, which is a sitemap index file pointing
> to http://tomcat.apache.org/sitemap-main.xml
> By doing this, you can emphasize more recent versions of docs. There are
> other benefits, but reducing poor Google search results (to me) is the
> biggest win.
> E.g. https://www.google.com/search?q=flink+reducingstate
> (search on flink
> reducing state) shows the 1.3 Javadocs (hit #1), master (1.6-SNAPSHOT)
> Javadocs (hit #2), and then many pages of other results.
> Whereas the Javadocs for 1.5
>
> and 1.4
>
> are nowhere to be found.
> {quote}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Closed] (FLINK-10670) Fix Correlate codegen error
[ https://issues.apache.org/jira/browse/FLINK-10670?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Till Rohrmann closed FLINK-10670. - Resolution: Fixed Fix Version/s: (was: 1.7.1) 1.7.0 Updated fix version > Fix Correlate codegen error > --- > > Key: FLINK-10670 > URL: https://issues.apache.org/jira/browse/FLINK-10670 > Project: Flink > Issue Type: Bug > Components: Table API SQL >Reporter: Ruidong Li >Assignee: Ruidong Li >Priority: Major > Labels: pull-request-available > Fix For: 1.5.6, 1.6.3, 1.7.0 > > > TableFunctionCollector should handle reuseInitCode and reuseMemberCode -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[GitHub] vtkstef commented on a change in pull request #6782: [FLINK-9083][Cassandra Connector] Add async backpressure support to Cassandra Connector
vtkstef commented on a change in pull request #6782: [FLINK-9083][Cassandra
Connector] Add async backpressure support to Cassandra Connector
URL: https://github.com/apache/flink/pull/6782#discussion_r234799715
##
File path:
flink-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraSinkBase.java
##
@@ -45,73 +48,63 @@
*/
public abstract class CassandraSinkBase extends RichSinkFunction
implements CheckpointedFunction {
protected final Logger log = LoggerFactory.getLogger(getClass());
+
+ // --- Cassandra Fields
---
+
+ private final ClusterBuilder builder;
+
protected transient Cluster cluster;
protected transient Session session;
- protected transient volatile Throwable exception;
- protected transient FutureCallback callback;
+ // Synchronization Fields
- private final ClusterBuilder builder;
+ private AtomicReference throwable;
+ private Semaphore semaphore;
+ private Phaser phaser;
Review comment:
Have you considered Guava's rate limiter? If so, I am curious, what made you
choose using Phaser and semaphore together instead of the rate limiter?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[jira] [Commented] (FLINK-9083) Add async backpressure support to Cassandra Connector
[
https://issues.apache.org/jira/browse/FLINK-9083?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692342#comment-16692342
]
ASF GitHub Bot commented on FLINK-9083:
---
vtkstef commented on a change in pull request #6782: [FLINK-9083][Cassandra
Connector] Add async backpressure support to Cassandra Connector
URL: https://github.com/apache/flink/pull/6782#discussion_r234799715
##
File path:
flink-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraSinkBase.java
##
@@ -45,73 +48,63 @@
*/
public abstract class CassandraSinkBase extends RichSinkFunction
implements CheckpointedFunction {
protected final Logger log = LoggerFactory.getLogger(getClass());
+
+ // --- Cassandra Fields
---
+
+ private final ClusterBuilder builder;
+
protected transient Cluster cluster;
protected transient Session session;
- protected transient volatile Throwable exception;
- protected transient FutureCallback callback;
+ // Synchronization Fields
- private final ClusterBuilder builder;
+ private AtomicReference throwable;
+ private Semaphore semaphore;
+ private Phaser phaser;
Review comment:
Have you considered Guava's rate limiter? If so, I am curious, what made you
choose using Phaser and semaphore together instead of the rate limiter?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Add async backpressure support to Cassandra Connector
> -
>
> Key: FLINK-9083
> URL: https://issues.apache.org/jira/browse/FLINK-9083
> Project: Flink
> Issue Type: Improvement
> Components: Cassandra Connector
>Reporter: Jacob Park
>Assignee: Jacob Park
>Priority: Minor
> Labels: pull-request-available
>
> As the CassandraSinkBase derivatives utilize async writes, they do not block
> the task to introduce any backpressure.
> I am currently using a semaphore to provide backpressure support by blocking
> at a maximum concurrent requests limit like how DataStax's Spark Cassandra
> Connector functions:
> [https://github.com/datastax/spark-cassandra-connector/blob/v2.0.7/spark-cassandra-connector/src/main/scala/com/datastax/spark/connector/writer/AsyncExecutor.scala#L18]
> This improvement has greatly improved the fault-tolerance of our Cassandra
> Sink Connector implementation on Apache Flink in production. I would like to
> contribute this feature back upstream.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Reopened] (FLINK-8997) Add sliding window aggregation to the job
[ https://issues.apache.org/jira/browse/FLINK-8997?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Till Rohrmann reopened FLINK-8997: -- > Add sliding window aggregation to the job > - > > Key: FLINK-8997 > URL: https://issues.apache.org/jira/browse/FLINK-8997 > Project: Flink > Issue Type: Sub-task > Components: Tests >Affects Versions: 1.5.0 >Reporter: Stefan Richter >Assignee: Dawid Wysakowicz >Priority: Major > Labels: pull-request-available > Fix For: 1.7.0 > > > The test job should also test windowing. Sliding windows are probably the > most demanding form, so this would be a good pick for the test. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Closed] (FLINK-10481) Wordcount end-to-end test in docker env unstable
[
https://issues.apache.org/jira/browse/FLINK-10481?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Till Rohrmann closed FLINK-10481.
-
Resolution: Fixed
Fix Version/s: (was: 1.7.1)
(was: 1.8.0)
1.7.0
Updated fix version
> Wordcount end-to-end test in docker env unstable
>
>
> Key: FLINK-10481
> URL: https://issues.apache.org/jira/browse/FLINK-10481
> Project: Flink
> Issue Type: Bug
> Components: Tests
>Affects Versions: 1.7.0
>Reporter: Till Rohrmann
>Assignee: Dawid Wysakowicz
>Priority: Critical
> Labels: pull-request-available, test-stability
> Fix For: 1.6.3, 1.7.0
>
>
> The {{Wordcount end-to-end test in docker env}} fails sometimes on Travis
> with the following problem:
> {code}
> Status: Downloaded newer image for java:8-jre-alpine
> ---> fdc893b19a14
> Step 2/16 : RUN apk add --no-cache bash snappy
> ---> [Warning] IPv4 forwarding is disabled. Networking will not work.
> ---> Running in 4329ebcd8a77
> fetch http://dl-cdn.alpinelinux.org/alpine/v3.4/main/x86_64/APKINDEX.tar.gz
> WARNING: Ignoring
> http://dl-cdn.alpinelinux.org/alpine/v3.4/main/x86_64/APKINDEX.tar.gz:
> temporary error (try again later)
> fetch
> http://dl-cdn.alpinelinux.org/alpine/v3.4/community/x86_64/APKINDEX.tar.gz
> WARNING: Ignoring
> http://dl-cdn.alpinelinux.org/alpine/v3.4/community/x86_64/APKINDEX.tar.gz:
> temporary error (try again later)
> ERROR: unsatisfiable constraints:
> bash (missing):
> required by: world[bash]
> snappy (missing):
> required by: world[snappy]
> The command '/bin/sh -c apk add --no-cache bash snappy' returned a non-zero
> code: 2
> {code}
> https://api.travis-ci.org/v3/job/434909395/log.txt
> It seems as if it is related to
> https://github.com/gliderlabs/docker-alpine/issues/264 and
> https://github.com/gliderlabs/docker-alpine/issues/279.
> We might want to switch to a different base image to avoid these problems in
> the future.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Closed] (FLINK-8997) Add sliding window aggregation to the job
[ https://issues.apache.org/jira/browse/FLINK-8997?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Till Rohrmann closed FLINK-8997. Resolution: Fixed Fix Version/s: (was: 1.7.1) (was: 1.8.0) 1.7.0 Updated fix version. > Add sliding window aggregation to the job > - > > Key: FLINK-8997 > URL: https://issues.apache.org/jira/browse/FLINK-8997 > Project: Flink > Issue Type: Sub-task > Components: Tests >Affects Versions: 1.5.0 >Reporter: Stefan Richter >Assignee: Dawid Wysakowicz >Priority: Major > Labels: pull-request-available > Fix For: 1.7.0 > > > The test job should also test windowing. Sliding windows are probably the > most demanding form, so this would be a good pick for the test. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Reopened] (FLINK-10481) Wordcount end-to-end test in docker env unstable
[
https://issues.apache.org/jira/browse/FLINK-10481?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Till Rohrmann reopened FLINK-10481:
---
> Wordcount end-to-end test in docker env unstable
>
>
> Key: FLINK-10481
> URL: https://issues.apache.org/jira/browse/FLINK-10481
> Project: Flink
> Issue Type: Bug
> Components: Tests
>Affects Versions: 1.7.0
>Reporter: Till Rohrmann
>Assignee: Dawid Wysakowicz
>Priority: Critical
> Labels: pull-request-available, test-stability
> Fix For: 1.6.3, 1.8.0, 1.7.1
>
>
> The {{Wordcount end-to-end test in docker env}} fails sometimes on Travis
> with the following problem:
> {code}
> Status: Downloaded newer image for java:8-jre-alpine
> ---> fdc893b19a14
> Step 2/16 : RUN apk add --no-cache bash snappy
> ---> [Warning] IPv4 forwarding is disabled. Networking will not work.
> ---> Running in 4329ebcd8a77
> fetch http://dl-cdn.alpinelinux.org/alpine/v3.4/main/x86_64/APKINDEX.tar.gz
> WARNING: Ignoring
> http://dl-cdn.alpinelinux.org/alpine/v3.4/main/x86_64/APKINDEX.tar.gz:
> temporary error (try again later)
> fetch
> http://dl-cdn.alpinelinux.org/alpine/v3.4/community/x86_64/APKINDEX.tar.gz
> WARNING: Ignoring
> http://dl-cdn.alpinelinux.org/alpine/v3.4/community/x86_64/APKINDEX.tar.gz:
> temporary error (try again later)
> ERROR: unsatisfiable constraints:
> bash (missing):
> required by: world[bash]
> snappy (missing):
> required by: world[snappy]
> The command '/bin/sh -c apk add --no-cache bash snappy' returned a non-zero
> code: 2
> {code}
> https://api.travis-ci.org/v3/job/434909395/log.txt
> It seems as if it is related to
> https://github.com/gliderlabs/docker-alpine/issues/264 and
> https://github.com/gliderlabs/docker-alpine/issues/279.
> We might want to switch to a different base image to avoid these problems in
> the future.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Reopened] (FLINK-10670) Fix Correlate codegen error
[ https://issues.apache.org/jira/browse/FLINK-10670?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Till Rohrmann reopened FLINK-10670: --- > Fix Correlate codegen error > --- > > Key: FLINK-10670 > URL: https://issues.apache.org/jira/browse/FLINK-10670 > Project: Flink > Issue Type: Bug > Components: Table API SQL >Reporter: Ruidong Li >Assignee: Ruidong Li >Priority: Major > Labels: pull-request-available > Fix For: 1.5.6, 1.6.3, 1.7.0 > > > TableFunctionCollector should handle reuseInitCode and reuseMemberCode -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Assigned] (FLINK-10934) Implement KubernetesJobClusterEntrypoint
[ https://issues.apache.org/jira/browse/FLINK-10934?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] JIN SUN reassigned FLINK-10934: --- Assignee: JIN SUN > Implement KubernetesJobClusterEntrypoint > > > Key: FLINK-10934 > URL: https://issues.apache.org/jira/browse/FLINK-10934 > Project: Flink > Issue Type: Sub-task >Reporter: JIN SUN >Assignee: JIN SUN >Priority: Major > > * Implement KubernetesJobClusterEntrypoint -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-10939) Add documents for Flink on native k8s
JIN SUN created FLINK-10939: --- Summary: Add documents for Flink on native k8s Key: FLINK-10939 URL: https://issues.apache.org/jira/browse/FLINK-10939 Project: Flink Issue Type: Sub-task Reporter: JIN SUN Assignee: JIN SUN -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-10938) Enable Flink on native k8s E2E Tests in Travis CI
JIN SUN created FLINK-10938: --- Summary: Enable Flink on native k8s E2E Tests in Travis CI Key: FLINK-10938 URL: https://issues.apache.org/jira/browse/FLINK-10938 Project: Flink Issue Type: Sub-task Reporter: JIN SUN Assignee: JIN SUN Add E2E tests to verify Flink on K8s integration -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-10937) Add scripts to create docker image for k8s
JIN SUN created FLINK-10937: --- Summary: Add scripts to create docker image for k8s Key: FLINK-10937 URL: https://issues.apache.org/jira/browse/FLINK-10937 Project: Flink Issue Type: Sub-task Reporter: JIN SUN Assignee: JIN SUN Add script to build docker image for flink on native k8s. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10932) Initial flink-kubernetes module with empty implementation
[ https://issues.apache.org/jira/browse/FLINK-10932?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692324#comment-16692324 ] ASF GitHub Bot commented on FLINK-10932: isunjin opened a new pull request #7144: [FLINK-10932] [ResourceManager] Initial flink-kubernetes module with empty implementation URL: https://github.com/apache/flink/pull/7144 ## What is the purpose of the change Initialize the skeleton module to start native k8s integration ## Brief change log - *Add flink-kubernetes module* - *Add first Implement of ClusterDescriptor* - *Add Interface of Kubernetes client* ## 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 ) - The S3 file system connector: ( no ) ## Documentation - Does this pull request introduce a new feature? (yes) - If yes, how is the feature documented? (the document will be introduced in other pull request) This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Initial flink-kubernetes module with empty implementation > - > > Key: FLINK-10932 > URL: https://issues.apache.org/jira/browse/FLINK-10932 > Project: Flink > Issue Type: Sub-task > Components: ResourceManager >Reporter: JIN SUN >Assignee: JIN SUN >Priority: Major > Labels: pull-request-available > > Initial the skeleton module to start native kubernetes integration. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-10936) Implement Command line tools
JIN SUN created FLINK-10936: --- Summary: Implement Command line tools Key: FLINK-10936 URL: https://issues.apache.org/jira/browse/FLINK-10936 Project: Flink Issue Type: Sub-task Reporter: JIN SUN Assignee: JIN SUN Implement command tools to start kubernetes sessions: * k8s-session.sh to start and stop a session like we did in yarn-session.sh * customized command line that will be invoked by CliFrontEnd and ./bin/flink run to submit job to kubernetes cluster -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Assigned] (FLINK-10935) Implement KubeClient with Faric8 Kubernetes clients
[ https://issues.apache.org/jira/browse/FLINK-10935?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] JIN SUN reassigned FLINK-10935: --- Assignee: JIN SUN > Implement KubeClient with Faric8 Kubernetes clients > > > Key: FLINK-10935 > URL: https://issues.apache.org/jira/browse/FLINK-10935 > Project: Flink > Issue Type: Sub-task >Reporter: JIN SUN >Assignee: JIN SUN >Priority: Major > > Implement KubeClient with Faric8 Kubernetes clients and add tests -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-10935) Implement KubeClient with Faric8 Kubernetes clients
JIN SUN created FLINK-10935: --- Summary: Implement KubeClient with Faric8 Kubernetes clients Key: FLINK-10935 URL: https://issues.apache.org/jira/browse/FLINK-10935 Project: Flink Issue Type: Sub-task Reporter: JIN SUN Implement KubeClient with Faric8 Kubernetes clients and add tests -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-10934) Implement KubernetesJobClusterEntrypoint
JIN SUN created FLINK-10934: --- Summary: Implement KubernetesJobClusterEntrypoint Key: FLINK-10934 URL: https://issues.apache.org/jira/browse/FLINK-10934 Project: Flink Issue Type: Sub-task Reporter: JIN SUN * Implement KubernetesJobClusterEntrypoint -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-10933) Implement KubernetesSessionClusterEntrypoint
JIN SUN created FLINK-10933: --- Summary: Implement KubernetesSessionClusterEntrypoint Key: FLINK-10933 URL: https://issues.apache.org/jira/browse/FLINK-10933 Project: Flink Issue Type: Sub-task Components: ResourceManager Reporter: JIN SUN Assignee: JIN SUN * Implement KubernetesSessionClusterEntrypoint * Implement TaskManager Entrypoint -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (FLINK-10932) Initial flink-kubernetes module with empty implementation
[ https://issues.apache.org/jira/browse/FLINK-10932?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] ASF GitHub Bot updated FLINK-10932: --- Labels: pull-request-available (was: ) > Initial flink-kubernetes module with empty implementation > - > > Key: FLINK-10932 > URL: https://issues.apache.org/jira/browse/FLINK-10932 > Project: Flink > Issue Type: Sub-task > Components: ResourceManager >Reporter: JIN SUN >Assignee: JIN SUN >Priority: Major > Labels: pull-request-available > > Initial the skeleton module to start native kubernetes integration. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[GitHub] isunjin opened a new pull request #7144: [FLINK-10932] [ResourceManager] Initial flink-kubernetes module with empty implementation
isunjin opened a new pull request #7144: [FLINK-10932] [ResourceManager] Initial flink-kubernetes module with empty implementation URL: https://github.com/apache/flink/pull/7144 ## What is the purpose of the change Initialize the skeleton module to start native k8s integration ## Brief change log - *Add flink-kubernetes module* - *Add first Implement of ClusterDescriptor* - *Add Interface of Kubernetes client* ## 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 ) - The S3 file system connector: ( no ) ## Documentation - Does this pull request introduce a new feature? (yes) - If yes, how is the feature documented? (the document will be introduced in other pull request) This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Resolved] (FLINK-10922) Refactor the placement of the Flink Kafka connector end to end test module
[ https://issues.apache.org/jira/browse/FLINK-10922?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Till Rohrmann resolved FLINK-10922. --- Resolution: Fixed Fixed via master: 33fbb42bb02c5490883c41d96a12ea5357ea7246 560986846c354bac80f0d6be3b75dc09741de87b b33e2c02044eef2cae8afd1b6c171da288fb4a63 1.7.0: 84cad5bda061787125cff60df7bcaf53b039c9c1 b992afa2ac43a0602e5bd8353da63d1539aab207 8c1bee6007376425fcf9915b79d2c646c0e9f071 > Refactor the placement of the Flink Kafka connector end to end test module > -- > > Key: FLINK-10922 > URL: https://issues.apache.org/jira/browse/FLINK-10922 > Project: Flink > Issue Type: Sub-task > Components: E2E Tests >Reporter: vinoyang >Assignee: vinoyang >Priority: Critical > Labels: pull-request-available > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10922) Refactor the placement of the Flink Kafka connector end to end test module
[
https://issues.apache.org/jira/browse/FLINK-10922?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692313#comment-16692313
]
ASF GitHub Bot commented on FLINK-10922:
asfgit closed pull request #7140: [FLINK-10922] Refactor the placement of the
Flink Kafka connector end to end test module
URL: https://github.com/apache/flink/pull/7140
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/flink-dist/src/main/assemblies/bin.xml
b/flink-dist/src/main/assemblies/bin.xml
index fa1e8048341..89228b39271 100644
--- a/flink-dist/src/main/assemblies/bin.xml
+++ b/flink-dist/src/main/assemblies/bin.xml
@@ -187,48 +187,6 @@ under the License.
-
-
-
../flink-examples/flink-examples-streaming-kafka/target
- examples/streaming
- 0644
-
- *.jar
-
-
-
flink-examples-streaming-kafka*.jar
- original-*.jar
-
-
-
-
-
-
../flink-examples/flink-examples-streaming-kafka-0.10/target
- examples/streaming
- 0644
-
- *.jar
-
-
-
flink-examples-streaming-kafka*.jar
- original-*.jar
-
-
-
-
-
-
../flink-examples/flink-examples-streaming-kafka-0.11/target
- examples/streaming
- 0644
-
- *.jar
-
-
-
flink-examples-streaming-kafka*.jar
- original-*.jar
-
-
-
../flink-libraries/flink-gelly-examples/target
diff --git a/flink-examples/flink-examples-streaming-kafka-base/pom.xml
b/flink-end-to-end-tests/flink-streaming-kafka-test-base/pom.xml
similarity index 88%
rename from flink-examples/flink-examples-streaming-kafka-base/pom.xml
rename to flink-end-to-end-tests/flink-streaming-kafka-test-base/pom.xml
index 463cd97adf6..1a86544dbe3 100644
--- a/flink-examples/flink-examples-streaming-kafka-base/pom.xml
+++ b/flink-end-to-end-tests/flink-streaming-kafka-test-base/pom.xml
@@ -21,14 +21,14 @@ under the License.
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance;
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd;>
- flink-examples
+ flink-end-to-end-tests
org.apache.flink
1.8-SNAPSHOT
4.0.0
-
flink-examples-streaming-kafka-base_${scala.binary.version}
- flink-examples-streaming-kafka-base
+
flink-streaming-kafka-test-base_${scala.binary.version}
+ flink-streaming-kafka-test-base
jar
diff --git
a/flink-examples/flink-examples-streaming-kafka-base/src/main/java/org/apache/flink/streaming/examples/kafka/base/CustomWatermarkExtractor.java
b/flink-end-to-end-tests/flink-streaming-kafka-test-base/src/main/java/org/apache/flink/streaming/kafka/test/base/CustomWatermarkExtractor.java
similarity index 97%
rename from
flink-examples/flink-examples-streaming-kafka-base/src/main/java/org/apache/flink/streaming/examples/kafka/base/CustomWatermarkExtractor.java
rename to
flink-end-to-end-tests/flink-streaming-kafka-test-base/src/main/java/org/apache/flink/streaming/kafka/test/base/CustomWatermarkExtractor.java
index 51de582dc0f..8b306ac9297 100644
---
a/flink-examples/flink-examples-streaming-kafka-base/src/main/java/org/apache/flink/streaming/examples/kafka/base/CustomWatermarkExtractor.java
+++
b/flink-end-to-end-tests/flink-streaming-kafka-test-base/src/main/java/org/apache/flink/streaming/kafka/test/base/CustomWatermarkExtractor.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.streaming.examples.kafka.base;
+package org.apache.flink.streaming.kafka.test.base;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import
org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
diff --git
[GitHub] asfgit closed pull request #7140: [FLINK-10922] Refactor the placement of the Flink Kafka connector end to end test module
asfgit closed pull request #7140: [FLINK-10922] Refactor the placement of the
Flink Kafka connector end to end test module
URL: https://github.com/apache/flink/pull/7140
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/flink-dist/src/main/assemblies/bin.xml
b/flink-dist/src/main/assemblies/bin.xml
index fa1e8048341..89228b39271 100644
--- a/flink-dist/src/main/assemblies/bin.xml
+++ b/flink-dist/src/main/assemblies/bin.xml
@@ -187,48 +187,6 @@ under the License.
-
-
-
../flink-examples/flink-examples-streaming-kafka/target
- examples/streaming
- 0644
-
- *.jar
-
-
-
flink-examples-streaming-kafka*.jar
- original-*.jar
-
-
-
-
-
-
../flink-examples/flink-examples-streaming-kafka-0.10/target
- examples/streaming
- 0644
-
- *.jar
-
-
-
flink-examples-streaming-kafka*.jar
- original-*.jar
-
-
-
-
-
-
../flink-examples/flink-examples-streaming-kafka-0.11/target
- examples/streaming
- 0644
-
- *.jar
-
-
-
flink-examples-streaming-kafka*.jar
- original-*.jar
-
-
-
../flink-libraries/flink-gelly-examples/target
diff --git a/flink-examples/flink-examples-streaming-kafka-base/pom.xml
b/flink-end-to-end-tests/flink-streaming-kafka-test-base/pom.xml
similarity index 88%
rename from flink-examples/flink-examples-streaming-kafka-base/pom.xml
rename to flink-end-to-end-tests/flink-streaming-kafka-test-base/pom.xml
index 463cd97adf6..1a86544dbe3 100644
--- a/flink-examples/flink-examples-streaming-kafka-base/pom.xml
+++ b/flink-end-to-end-tests/flink-streaming-kafka-test-base/pom.xml
@@ -21,14 +21,14 @@ under the License.
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance;
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd;>
- flink-examples
+ flink-end-to-end-tests
org.apache.flink
1.8-SNAPSHOT
4.0.0
-
flink-examples-streaming-kafka-base_${scala.binary.version}
- flink-examples-streaming-kafka-base
+
flink-streaming-kafka-test-base_${scala.binary.version}
+ flink-streaming-kafka-test-base
jar
diff --git
a/flink-examples/flink-examples-streaming-kafka-base/src/main/java/org/apache/flink/streaming/examples/kafka/base/CustomWatermarkExtractor.java
b/flink-end-to-end-tests/flink-streaming-kafka-test-base/src/main/java/org/apache/flink/streaming/kafka/test/base/CustomWatermarkExtractor.java
similarity index 97%
rename from
flink-examples/flink-examples-streaming-kafka-base/src/main/java/org/apache/flink/streaming/examples/kafka/base/CustomWatermarkExtractor.java
rename to
flink-end-to-end-tests/flink-streaming-kafka-test-base/src/main/java/org/apache/flink/streaming/kafka/test/base/CustomWatermarkExtractor.java
index 51de582dc0f..8b306ac9297 100644
---
a/flink-examples/flink-examples-streaming-kafka-base/src/main/java/org/apache/flink/streaming/examples/kafka/base/CustomWatermarkExtractor.java
+++
b/flink-end-to-end-tests/flink-streaming-kafka-test-base/src/main/java/org/apache/flink/streaming/kafka/test/base/CustomWatermarkExtractor.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.streaming.examples.kafka.base;
+package org.apache.flink.streaming.kafka.test.base;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import
org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
diff --git
a/flink-examples/flink-examples-streaming-kafka-base/src/main/java/org/apache/flink/streaming/examples/kafka/base/KafkaEvent.java
b/flink-end-to-end-tests/flink-streaming-kafka-test-base/src/main/java/org/apache/flink/streaming/kafka/test/base/KafkaEvent.java
similarity index 97%
rename from
[jira] [Commented] (FLINK-9541) Add robots.txt and sitemap.xml to Flink website
[
https://issues.apache.org/jira/browse/FLINK-9541?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16692292#comment-16692292
]
Ken Krugler commented on FLINK-9541:
Hi [~fhueske] - thanks for the input, some comments below.
* In order to avoid crawling flink-docs-master, we'd have to add a Disallow
entry to the robots.txt file at ci.apache.org. Which is doable, yes, though I
imagine we might get some push-back from the ops team to avoid having lots of
groups editing the same file.
* Yes, we could update the sitemap to adjust weights with each new release.
I'm wondering about the easiest way to build the sitemap as part of the release
process, as it needs both the (rendered) markdown files and the generated
JavaDoc files.
* At some point you'll want to remove older versions of the documents. But
anyone who isn't upgrading will then have problems, so I imagine you want to be
pretty conservative. E.g. currently you've got the 1.0 - 1.3 versions, but
seems like everyone should be on 1.4 at least, yes?
* And to cover [~Zentol]'s point, it would be better to figure out how (if
possible) to have the httpd server return the canonical link info for the
current release, so that you don't have to edit all of the older release files
(and this would work for JavaDocs too).
> Add robots.txt and sitemap.xml to Flink website
> ---
>
> Key: FLINK-9541
> URL: https://issues.apache.org/jira/browse/FLINK-9541
> Project: Flink
> Issue Type: Improvement
> Components: Project Website
>Reporter: Fabian Hueske
>Priority: Major
>
> From the [dev mailing
> list|https://lists.apache.org/thread.html/71ce1bfbed1cf5f0069b27a46df1cd4dccbe8abefa75ac85601b088b@%3Cdev.flink.apache.org%3E]:
> {quote}
> It would help to add a sitemap (and the robots.txt required to reference it)
> for flink.apache.org and ci.apache.org (for /projects/flink)
> You can see what Tomcat did along these lines -
> http://tomcat.apache.org/robots.txt references
> http://tomcat.apache.org/sitemap.xml, which is a sitemap index file pointing
> to http://tomcat.apache.org/sitemap-main.xml
> By doing this, you can emphasize more recent versions of docs. There are
> other benefits, but reducing poor Google search results (to me) is the
> biggest win.
> E.g. https://www.google.com/search?q=flink+reducingstate
> (search on flink
> reducing state) shows the 1.3 Javadocs (hit #1), master (1.6-SNAPSHOT)
> Javadocs (hit #2), and then many pages of other results.
> Whereas the Javadocs for 1.5
>
> and 1.4
>
> are nowhere to be found.
> {quote}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Resolved] (FLINK-10893) Streaming File Sink s3 end-to-end test failed on travis
[
https://issues.apache.org/jira/browse/FLINK-10893?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Stephan Ewen resolved FLINK-10893.
--
Resolution: Fixed
Fix Version/s: (was: 1.8.0)
1.7.0
Fixed in
- 1.7 via 65ef428fd0ec3ee56071f50f26f3221ade1a26ae
- 1.8 via b7bbd06cd6c6875d1103f393d46b530a788a18a1
> Streaming File Sink s3 end-to-end test failed on travis
> ---
>
> Key: FLINK-10893
> URL: https://issues.apache.org/jira/browse/FLINK-10893
> Project: Flink
> Issue Type: Bug
> Components: E2E Tests
>Reporter: Dawid Wysakowicz
>Assignee: Stephan Ewen
>Priority: Major
> Fix For: 1.7.0
>
>
> https://api.travis-ci.org/v3/job/454950368/log.txt
> Test failed with exception:
> {code}
> Exception in thread "main" com.amazonaws.services.s3.model.AmazonS3Exception:
> The bucket is in this region: us-east-1. Please use this region to retry the
> request (Service: Amazon S3; Status Code: 301; Error Code: PermanentRedirect;
> Request ID: 7C554BA6EB1DDCAE; S3 Extended Request ID:
> hPd88Lpqbh1wPV83GN3YgGH3h3e9ct5pDvhoFrTI3JGipfrCoztr/bWvHCUE9l7D9fBTXJMkvus=),
> S3 Extended Request ID:
> hPd88Lpqbh1wPV83GN3YgGH3h3e9ct5pDvhoFrTI3JGipfrCoztr/bWvHCUE9l7D9fBTXJMkvus=
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutor.handleErrorResponse(AmazonHttpClient.java:1695)
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeOneRequest(AmazonHttpClient.java:1350)
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeHelper(AmazonHttpClient.java:1101)
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutor.doExecute(AmazonHttpClient.java:758)
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeWithTimer(AmazonHttpClient.java:732)
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutor.execute(AmazonHttpClient.java:714)
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutor.access$500(AmazonHttpClient.java:674)
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutionBuilderImpl.execute(AmazonHttpClient.java:656)
> at
> com.amazonaws.http.AmazonHttpClient.execute(AmazonHttpClient.java:520)
> at
> com.amazonaws.services.s3.AmazonS3Client.invoke(AmazonS3Client.java:4443)
> at
> com.amazonaws.services.s3.AmazonS3Client.invoke(AmazonS3Client.java:4390)
> at
> com.amazonaws.services.s3.AmazonS3Client.invoke(AmazonS3Client.java:4384)
> at
> com.amazonaws.services.s3.AmazonS3Client.listObjects(AmazonS3Client.java:844)
> at
> com.amazonaws.services.s3.AmazonS3Client.listObjects(AmazonS3Client.java:818)
> at
> org.apache.flink.streaming.tests.util.s3.S3UtilProgram.listByFullPathPrefix(S3UtilProgram.java:107)
> at
> org.apache.flink.streaming.tests.util.s3.S3UtilProgram.deleteByFullPathPrefix(S3UtilProgram.java:164)
> at
> org.apache.flink.streaming.tests.util.s3.S3UtilProgram.main(S3UtilProgram.java:97)
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Closed] (FLINK-10893) Streaming File Sink s3 end-to-end test failed on travis
[
https://issues.apache.org/jira/browse/FLINK-10893?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Stephan Ewen closed FLINK-10893.
> Streaming File Sink s3 end-to-end test failed on travis
> ---
>
> Key: FLINK-10893
> URL: https://issues.apache.org/jira/browse/FLINK-10893
> Project: Flink
> Issue Type: Bug
> Components: E2E Tests
>Reporter: Dawid Wysakowicz
>Assignee: Stephan Ewen
>Priority: Major
> Fix For: 1.7.0
>
>
> https://api.travis-ci.org/v3/job/454950368/log.txt
> Test failed with exception:
> {code}
> Exception in thread "main" com.amazonaws.services.s3.model.AmazonS3Exception:
> The bucket is in this region: us-east-1. Please use this region to retry the
> request (Service: Amazon S3; Status Code: 301; Error Code: PermanentRedirect;
> Request ID: 7C554BA6EB1DDCAE; S3 Extended Request ID:
> hPd88Lpqbh1wPV83GN3YgGH3h3e9ct5pDvhoFrTI3JGipfrCoztr/bWvHCUE9l7D9fBTXJMkvus=),
> S3 Extended Request ID:
> hPd88Lpqbh1wPV83GN3YgGH3h3e9ct5pDvhoFrTI3JGipfrCoztr/bWvHCUE9l7D9fBTXJMkvus=
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutor.handleErrorResponse(AmazonHttpClient.java:1695)
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeOneRequest(AmazonHttpClient.java:1350)
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeHelper(AmazonHttpClient.java:1101)
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutor.doExecute(AmazonHttpClient.java:758)
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeWithTimer(AmazonHttpClient.java:732)
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutor.execute(AmazonHttpClient.java:714)
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutor.access$500(AmazonHttpClient.java:674)
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutionBuilderImpl.execute(AmazonHttpClient.java:656)
> at
> com.amazonaws.http.AmazonHttpClient.execute(AmazonHttpClient.java:520)
> at
> com.amazonaws.services.s3.AmazonS3Client.invoke(AmazonS3Client.java:4443)
> at
> com.amazonaws.services.s3.AmazonS3Client.invoke(AmazonS3Client.java:4390)
> at
> com.amazonaws.services.s3.AmazonS3Client.invoke(AmazonS3Client.java:4384)
> at
> com.amazonaws.services.s3.AmazonS3Client.listObjects(AmazonS3Client.java:844)
> at
> com.amazonaws.services.s3.AmazonS3Client.listObjects(AmazonS3Client.java:818)
> at
> org.apache.flink.streaming.tests.util.s3.S3UtilProgram.listByFullPathPrefix(S3UtilProgram.java:107)
> at
> org.apache.flink.streaming.tests.util.s3.S3UtilProgram.deleteByFullPathPrefix(S3UtilProgram.java:164)
> at
> org.apache.flink.streaming.tests.util.s3.S3UtilProgram.main(S3UtilProgram.java:97)
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Assigned] (FLINK-10893) Streaming File Sink s3 end-to-end test failed on travis
[
https://issues.apache.org/jira/browse/FLINK-10893?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Stephan Ewen reassigned FLINK-10893:
Assignee: Stephan Ewen (was: Andrey Zagrebin)
> Streaming File Sink s3 end-to-end test failed on travis
> ---
>
> Key: FLINK-10893
> URL: https://issues.apache.org/jira/browse/FLINK-10893
> Project: Flink
> Issue Type: Bug
> Components: E2E Tests
>Reporter: Dawid Wysakowicz
>Assignee: Stephan Ewen
>Priority: Major
> Fix For: 1.8.0
>
>
> https://api.travis-ci.org/v3/job/454950368/log.txt
> Test failed with exception:
> {code}
> Exception in thread "main" com.amazonaws.services.s3.model.AmazonS3Exception:
> The bucket is in this region: us-east-1. Please use this region to retry the
> request (Service: Amazon S3; Status Code: 301; Error Code: PermanentRedirect;
> Request ID: 7C554BA6EB1DDCAE; S3 Extended Request ID:
> hPd88Lpqbh1wPV83GN3YgGH3h3e9ct5pDvhoFrTI3JGipfrCoztr/bWvHCUE9l7D9fBTXJMkvus=),
> S3 Extended Request ID:
> hPd88Lpqbh1wPV83GN3YgGH3h3e9ct5pDvhoFrTI3JGipfrCoztr/bWvHCUE9l7D9fBTXJMkvus=
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutor.handleErrorResponse(AmazonHttpClient.java:1695)
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeOneRequest(AmazonHttpClient.java:1350)
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeHelper(AmazonHttpClient.java:1101)
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutor.doExecute(AmazonHttpClient.java:758)
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeWithTimer(AmazonHttpClient.java:732)
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutor.execute(AmazonHttpClient.java:714)
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutor.access$500(AmazonHttpClient.java:674)
> at
> com.amazonaws.http.AmazonHttpClient$RequestExecutionBuilderImpl.execute(AmazonHttpClient.java:656)
> at
> com.amazonaws.http.AmazonHttpClient.execute(AmazonHttpClient.java:520)
> at
> com.amazonaws.services.s3.AmazonS3Client.invoke(AmazonS3Client.java:4443)
> at
> com.amazonaws.services.s3.AmazonS3Client.invoke(AmazonS3Client.java:4390)
> at
> com.amazonaws.services.s3.AmazonS3Client.invoke(AmazonS3Client.java:4384)
> at
> com.amazonaws.services.s3.AmazonS3Client.listObjects(AmazonS3Client.java:844)
> at
> com.amazonaws.services.s3.AmazonS3Client.listObjects(AmazonS3Client.java:818)
> at
> org.apache.flink.streaming.tests.util.s3.S3UtilProgram.listByFullPathPrefix(S3UtilProgram.java:107)
> at
> org.apache.flink.streaming.tests.util.s3.S3UtilProgram.deleteByFullPathPrefix(S3UtilProgram.java:164)
> at
> org.apache.flink.streaming.tests.util.s3.S3UtilProgram.main(S3UtilProgram.java:97)
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Assigned] (FLINK-9495) Implement ResourceManager for Kubernetes
[ https://issues.apache.org/jira/browse/FLINK-9495?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] JIN SUN reassigned FLINK-9495: -- Assignee: JIN SUN > Implement ResourceManager for Kubernetes > > > Key: FLINK-9495 > URL: https://issues.apache.org/jira/browse/FLINK-9495 > Project: Flink > Issue Type: Sub-task > Components: ResourceManager >Affects Versions: 1.5.0 >Reporter: Elias Levy >Assignee: JIN SUN >Priority: Major > > I noticed there is no issue for developing a Kubernetes specific > ResourceManager under FLIP-6, so I am creating this issue. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-10932) Initial flink-kubernetes module with empty implementation
JIN SUN created FLINK-10932: --- Summary: Initial flink-kubernetes module with empty implementation Key: FLINK-10932 URL: https://issues.apache.org/jira/browse/FLINK-10932 Project: Flink Issue Type: Sub-task Components: ResourceManager Reporter: JIN SUN Assignee: JIN SUN Initial the skeleton module to start native kubernetes integration. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Assigned] (FLINK-9955) Kubernetes ClusterDescriptor
[
https://issues.apache.org/jira/browse/FLINK-9955?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
JIN SUN reassigned FLINK-9955:
--
Assignee: JIN SUN
> Kubernetes ClusterDescriptor
>
>
> Key: FLINK-9955
> URL: https://issues.apache.org/jira/browse/FLINK-9955
> Project: Flink
> Issue Type: Sub-task
> Components: Client
>Reporter: Till Rohrmann
>Assignee: JIN SUN
>Priority: Major
> Fix For: 1.8.0
>
>
> In order to start programmatically a Flink cluster on Kubernetes, we need a
> {{KubernetesClusterDescriptor}} implementation.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Closed] (FLINK-6022) Don't serialise Schema when serialising Avro GenericRecord
[ https://issues.apache.org/jira/browse/FLINK-6022?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Stephan Ewen closed FLINK-6022. --- Resolution: Fixed Fix Version/s: (was: 1.8.0) This has been fixed a few versions back with the introduction of a proper AvroTypeInfo > Don't serialise Schema when serialising Avro GenericRecord > -- > > Key: FLINK-6022 > URL: https://issues.apache.org/jira/browse/FLINK-6022 > Project: Flink > Issue Type: Improvement > Components: Type Serialization System >Reporter: Robert Metzger >Assignee: Stephan Ewen >Priority: Major > > Currently, Flink is serializing the schema for each Avro GenericRecord in the > stream. > This leads to a lot of overhead over the wire/disk + high serialization costs. > Therefore, I'm proposing to improve the support for GenericRecord in Flink by > shipping the schema to each serializer through the AvroTypeInformation. > Then, we can only support GenericRecords with the same type per stream, but > the performance will be much better. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
