[jira] [Updated] (FLINK-17875) Support state TTL for remote functions
[ https://issues.apache.org/jira/browse/FLINK-17875?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] ASF GitHub Bot updated FLINK-17875: --- Labels: pull-request-available (was: ) > Support state TTL for remote functions > -- > > Key: FLINK-17875 > URL: https://issues.apache.org/jira/browse/FLINK-17875 > Project: Flink > Issue Type: Task > Components: Stateful Functions >Reporter: Tzu-Li (Gordon) Tai >Assignee: Tzu-Li (Gordon) Tai >Priority: Blocker > Labels: pull-request-available > Fix For: statefun-2.1.0 > > > With FLINK-17644, we now have support for embedded functions. > This should be extended to remote functions, by allowing the module specs to > define the TTL for declared remote function state. > With this, it is also likely that we need to uptick the version for the YAML > module spec. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink-statefun] tzulitai opened a new pull request #116: [FLINK-17875] Initial support of state TTL for remote functions
tzulitai opened a new pull request #116: URL: https://github.com/apache/flink-statefun/pull/116 This is an initial support for state TTL for remote functions. Users define state TTL in their YAML modules like so: ``` functions: - function: states: - name: ttlDurationMillis: 6 # optional key ``` The current implementation has some limitations due to how we are multiplexing remote function's user state in a single `PersistedTable` (see FLINK-17954): - The actual TTL being set will be the longest duration across all registered state - The only supported expiration mode now is `AFTER_READ_AND_WRITE`. That can be added as a `ttlExpireMode` key in the YAML spec in the future. ## Main change log - 8557fba to d27b143 some pre-req refactoring of `JsonModule` and `HttpFunctionSpec`. This refactors out version-specific JSON parsing logic to a separate class (`JsonModuleSpecParser`), and any concerns such as default values for optional keys, to `HttpFunctionSpec`. - fcc25ec to 91e22ef Introduce a `StateSpec` model class to `HttpFunctionSpec` that captures the configured state ttl duration for each state. The `StateSpec` is then used in `RequestReplyFunction` to correctly set the state TTL on the state handles. - 39a8745 to 0f7eb9a Adds a spec parser for v2.0 format, and wires it in to allow the version to be recognized. - f48c64e Refactors `JsonModuleTest` so that it runs tests for both v1.0 and v2.0 format ## Verifying - New `JsonModuleTest` covers basic parsing of the new format - End-to-end tests don't fail imply that remote function state access is not broken due to this change - Manually adapted the Python greeter example to check that state is expired. This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] becketqin commented on pull request #12306: [FLINK-17899][runtime] Integrate FLIP-126 Watermarks with FLIP-27 Sources
becketqin commented on pull request #12306: URL: https://github.com/apache/flink/pull/12306#issuecomment-63121 Thanks for updating the patch and fix the bugs. +1 to merge the PR. This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[jira] [Commented] (FLINK-17953) OverWindow doesn't support to order by non-time attribute in batch mode for Table API program
[
https://issues.apache.org/jira/browse/FLINK-17953?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17117407#comment-17117407
]
godfrey he commented on FLINK-17953:
blink planner does not require the order filed must be time attribute type for
batch sql. I think we can remove the validation and let the specific planner do
the validation.
> OverWindow doesn't support to order by non-time attribute in batch mode for
> Table API program
> -
>
> Key: FLINK-17953
> URL: https://issues.apache.org/jira/browse/FLINK-17953
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.9.0, 1.10.0, 1.11.0
>Reporter: Dian Fu
>Priority: Major
>
> For a simple batch job tested in blink planner:
> {code:java}
> INSERT INTO results
> SELECT id, sum(sales)
> OVER (PARTITION BY id ORDER BY ts ROWS BETWEEN 2 PRECEDING AND 0 FOLLOWING)
> FROM input
> {code}
> It could pass if written in SQL. However, if we rewrite it in Table API, it
> will throw the following exception:
> {code:java}
> py4j.protocol.Py4JJavaError: An error occurred while calling
> o85.select.py4j.protocol.Py4JJavaError: An error occurred while calling
> o85.select.: org.apache.flink.table.api.ValidationException: Ordering must be
> defined on a time attribute. at
> org.apache.flink.table.planner.expressions.PlannerTypeInferenceUtilImpl.validateArguments(PlannerTypeInferenceUtilImpl.java:112)
> at
> org.apache.flink.table.planner.expressions.PlannerTypeInferenceUtilImpl.runTypeInference(PlannerTypeInferenceUtilImpl.java:71)
> at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.runLegacyTypeInference(ResolveCallByArgumentsRule.java:218)
> at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.lambda$visit$2(ResolveCallByArgumentsRule.java:134)
> at java.util.Optional.orElseGet(Optional.java:267) at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.visit(ResolveCallByArgumentsRule.java:134)
> at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.visit(ResolveCallByArgumentsRule.java:89)
> at
> org.apache.flink.table.expressions.ApiExpressionVisitor.visit(ApiExpressionVisitor.java:39)
> at
> org.apache.flink.table.expressions.UnresolvedCallExpression.accept(UnresolvedCallExpression.java:132)
> at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.visit(ResolveCallByArgumentsRule.java:124)
> at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.visit(ResolveCallByArgumentsRule.java:89)
> at
> org.apache.flink.table.expressions.ApiExpressionVisitor.visit(ApiExpressionVisitor.java:39)
> at
> org.apache.flink.table.expressions.UnresolvedCallExpression.accept(UnresolvedCallExpression.java:132)
> at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule.lambda$apply$0(ResolveCallByArgumentsRule.java:83)
> at java.util.stream.ReferencePipeline$7$1.accept(ReferencePipeline.java:267)
> at
> java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1374)
> at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481) at
> java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
> at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708)
> at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) at
> java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:499) at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule.apply(ResolveCallByArgumentsRule.java:84)
> at
> org.apache.flink.table.expressions.resolver.ExpressionResolver.lambda$null$1(ExpressionResolver.java:211)
> at java.util.function.Function.lambda$andThen$1(Function.java:88) at
> org.apache.flink.table.expressions.resolver.ExpressionResolver.resolve(ExpressionResolver.java:178)
> at
> org.apache.flink.table.operations.utils.OperationTreeBuilder.projectInternal(OperationTreeBuilder.java:191)
> at
> org.apache.flink.table.operations.utils.OperationTreeBuilder.project(OperationTreeBuilder.java:170)
> at
> org.apache.flink.table.api.internal.TableImpl$OverWindowedTableImpl.select(TableImpl.java:953)
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:498) at
>
[GitHub] [flink] flinkbot edited a comment on pull request #12323: [FLINK-17925][fs-connector] Fix Filesystem options to default values and types
flinkbot edited a comment on pull request #12323: URL: https://github.com/apache/flink/pull/12323#issuecomment-633592742 ## CI report: * 32ca1d9dfc0cb2ab212c123240de70d084f7bc12 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2184) * 5183374cf406535be6ba4c8a4de100e88a0d658c UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[jira] [Issue Comment Deleted] (FLINK-17953) OverWindow doesn't support to order by non-time attribute in batch mode for Table API program
[
https://issues.apache.org/jira/browse/FLINK-17953?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Dian Fu updated FLINK-17953:
Comment: was deleted
(was: In streaming jobs, there are also the same kind of problems if the source
is defined through DDL.
For the following job:
{code}
source_ddl = """
create table input(
id INT,
sales FLOAT,
word VARCHAR,
ts TIMESTAMP(3),
WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
) with (
'connector' = 'filesystem',
'format' = 'csv',
'path' = '{}'
)
""".format(input_data_path)
t_env.execute_sql(source_ddl)
sink_ddl = """
create table results(
id INT,
total_sales FLOAT
) with (
'connector' = 'filesystem',
'format' = 'csv',
'path' = '{}'
)
""".format(result_data_path)
t_env.execute_sql(sink_ddl)
table = t_env.from_path("input").over_window(
Over.partition_by("id").order_by("ts").preceding("2.rows").following("current_row").alias('w'))
\
.select("id, sum(sales) over w as total_sales")
{code}
It will also throw the same kind of exception as described in this JIRA.)
> OverWindow doesn't support to order by non-time attribute in batch mode for
> Table API program
> -
>
> Key: FLINK-17953
> URL: https://issues.apache.org/jira/browse/FLINK-17953
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.9.0, 1.10.0, 1.11.0
>Reporter: Dian Fu
>Priority: Major
>
> For a simple batch job tested in blink planner:
> {code:java}
> INSERT INTO results
> SELECT id, sum(sales)
> OVER (PARTITION BY id ORDER BY ts ROWS BETWEEN 2 PRECEDING AND 0 FOLLOWING)
> FROM input
> {code}
> It could pass if written in SQL. However, if we rewrite it in Table API, it
> will throw the following exception:
> {code:java}
> py4j.protocol.Py4JJavaError: An error occurred while calling
> o85.select.py4j.protocol.Py4JJavaError: An error occurred while calling
> o85.select.: org.apache.flink.table.api.ValidationException: Ordering must be
> defined on a time attribute. at
> org.apache.flink.table.planner.expressions.PlannerTypeInferenceUtilImpl.validateArguments(PlannerTypeInferenceUtilImpl.java:112)
> at
> org.apache.flink.table.planner.expressions.PlannerTypeInferenceUtilImpl.runTypeInference(PlannerTypeInferenceUtilImpl.java:71)
> at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.runLegacyTypeInference(ResolveCallByArgumentsRule.java:218)
> at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.lambda$visit$2(ResolveCallByArgumentsRule.java:134)
> at java.util.Optional.orElseGet(Optional.java:267) at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.visit(ResolveCallByArgumentsRule.java:134)
> at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.visit(ResolveCallByArgumentsRule.java:89)
> at
> org.apache.flink.table.expressions.ApiExpressionVisitor.visit(ApiExpressionVisitor.java:39)
> at
> org.apache.flink.table.expressions.UnresolvedCallExpression.accept(UnresolvedCallExpression.java:132)
> at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.visit(ResolveCallByArgumentsRule.java:124)
> at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.visit(ResolveCallByArgumentsRule.java:89)
> at
> org.apache.flink.table.expressions.ApiExpressionVisitor.visit(ApiExpressionVisitor.java:39)
> at
> org.apache.flink.table.expressions.UnresolvedCallExpression.accept(UnresolvedCallExpression.java:132)
> at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule.lambda$apply$0(ResolveCallByArgumentsRule.java:83)
> at java.util.stream.ReferencePipeline$7$1.accept(ReferencePipeline.java:267)
> at
> java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1374)
> at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481) at
> java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
> at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708)
> at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) at
> java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:499) at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule.apply(ResolveCallByArgumentsRule.java:84)
> at
> org.apache.flink.table.expressions.resolver.ExpressionResolver.lambda$null$1(ExpressionResolver.java:211)
> at java.util.function.Function.lambda$andThen$1(Function.java:88) at
> org.apache.flink.table.expressions.resolver.ExpressionResolver.resolve(ExpressionResolver.java:178)
>
[GitHub] [flink-web] zjffdu commented on pull request #339: [blog] flink on zeppelin
zjffdu commented on pull request #339: URL: https://github.com/apache/flink-web/pull/339#issuecomment-634440256 @MarkSfik @morsapaes Thanks for the review, I have updated the post This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] flinkbot edited a comment on pull request #12334: [FLINK-17934][FLINK-17955][fs-connector] StreamingFileWriter should set chainingStrategy & BucketLifeCycleListener should just in Buc
flinkbot edited a comment on pull request #12334: URL: https://github.com/apache/flink/pull/12334#issuecomment-633879796 ## CI report: * e3f49ff243a1ef02b1fe3caeb4001e42d2c38636 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2242) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[jira] [Comment Edited] (FLINK-17581) Update translation of S3 documentation
[ https://issues.apache.org/jira/browse/FLINK-17581?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17117402#comment-17117402 ] Cedric Chen edited comment on FLINK-17581 at 5/27/20, 5:19 AM: --- Hi [~rmetzger] , I'm a starter and willing to do it! was (Author: wydhcws): Hi [~rmetzger] , I'm a starter and I am willing to do it! > Update translation of S3 documentation > -- > > Key: FLINK-17581 > URL: https://issues.apache.org/jira/browse/FLINK-17581 > Project: Flink > Issue Type: Task > Components: chinese-translation, Documentation >Reporter: Robert Metzger >Priority: Major > > The change in > https://github.com/apache/flink/commit/7c5ac3584e42a0e7ebc5e78c532887bf4d383d9d > needs to be added to the Chinese variant of the documentation page. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (FLINK-17581) Update translation of S3 documentation
[ https://issues.apache.org/jira/browse/FLINK-17581?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17117402#comment-17117402 ] Nuo Chen commented on FLINK-17581: -- Hi [~rmetzger] , I'm a starter and I am willing to do it! > Update translation of S3 documentation > -- > > Key: FLINK-17581 > URL: https://issues.apache.org/jira/browse/FLINK-17581 > Project: Flink > Issue Type: Task > Components: chinese-translation, Documentation >Reporter: Robert Metzger >Priority: Major > > The change in > https://github.com/apache/flink/commit/7c5ac3584e42a0e7ebc5e78c532887bf4d383d9d > needs to be added to the Chinese variant of the documentation page. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] JingsongLi commented on a change in pull request #12323: [FLINK-17925][fs-connector] Fix Filesystem options to default values and types
JingsongLi commented on a change in pull request #12323:
URL: https://github.com/apache/flink/pull/12323#discussion_r430861320
##
File path:
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemOptions.java
##
@@ -132,12 +133,12 @@
public static final ConfigOption SINK_PARTITION_COMMIT_TRIGGER =
key("sink.partition-commit.trigger")
.stringType()
- .defaultValue("partition-time")
+ .defaultValue("process-time")
.withDescription("Trigger type for
partition commit:" +
Review comment:
```
'process-time': use processing time, if 'current time' > 'partition creation
time' + 'delay', will commit the partition.
'partition-time': extract time from partition, if 'current event time' >
'time from partition values' + 'delay', will commit the partition.
```
`time from partition values` is a kind of event time.
`partition creation time` is from processing time.
So they're not orthogonal.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to 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
[GitHub] [flink] JingsongLi commented on a change in pull request #12323: [FLINK-17925][fs-connector] Fix Filesystem options to default values and types
JingsongLi commented on a change in pull request #12323:
URL: https://github.com/apache/flink/pull/12323#discussion_r430861320
##
File path:
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemOptions.java
##
@@ -132,12 +133,12 @@
public static final ConfigOption SINK_PARTITION_COMMIT_TRIGGER =
key("sink.partition-commit.trigger")
.stringType()
- .defaultValue("partition-time")
+ .defaultValue("process-time")
.withDescription("Trigger type for
partition commit:" +
Review comment:
```
'process-time': use processing time, if 'current time' > 'partition creation
time' + 'delay', will commit the partition.
'partition-time': extract time from partition, if 'current event time' >
'time from partition values' + 'delay', will commit the partition.
```
`time from partition values` is a kind of event time.
`partition creation time` is from processing time.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to 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
[GitHub] [flink] JingsongLi commented on a change in pull request #12323: [FLINK-17925][fs-connector] Fix Filesystem options to default values and types
JingsongLi commented on a change in pull request #12323:
URL: https://github.com/apache/flink/pull/12323#discussion_r430859922
##
File path:
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/PartitionCommitPolicy.java
##
@@ -121,4 +122,20 @@
}
}).collect(Collectors.toList());
}
+
+ /**
+* Validate commit policy.
+*/
+ static void validatePolicyChain(boolean isEmptyMetastore, String
policyKind) {
+ if (policyKind != null) {
+ String[] policyStrings = policyKind.split(",");
+ for (String policy : policyStrings) {
+ if (isEmptyMetastore &&
METASTORE.equalsIgnoreCase(policy)) {
+ throw new ValidationException(
+ "Can not configure a
metastore partition commit policy for" +
+ " a
table without metastore.");
Review comment:
I think we can just say `file system connector`.
`Can not configure a 'metastore' partition commit policy for a file system
table. You can only configure 'metastore' partition commit policy for a hive
table.`
This is an automated message from the Apache Git Service.
To respond to the message, please log on to 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
[GitHub] [flink] JingsongLi commented on a change in pull request #12323: [FLINK-17925][fs-connector] Fix Filesystem options to default values and types
JingsongLi commented on a change in pull request #12323:
URL: https://github.com/apache/flink/pull/12323#discussion_r430859222
##
File path:
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/EmptyMetaStoreFactory.java
##
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.filesystem;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.core.fs.Path;
+
+import java.util.LinkedHashMap;
+import java.util.Optional;
+
+/**
+ * Empty implementation {@link TableMetaStoreFactory}.
Review comment:
I will deprecate `getLocationPath`. It should be empty implementation.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to 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
[GitHub] [flink] JingsongLi commented on a change in pull request #12323: [FLINK-17925][fs-connector] Fix Filesystem options to default values and types
JingsongLi commented on a change in pull request #12323:
URL: https://github.com/apache/flink/pull/12323#discussion_r430859222
##
File path:
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/EmptyMetaStoreFactory.java
##
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.filesystem;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.core.fs.Path;
+
+import java.util.LinkedHashMap;
+import java.util.Optional;
+
+/**
+ * Empty implementation {@link TableMetaStoreFactory}.
Review comment:
I will deprecate `getLocationPath`(Only deprecated batch sink use this).
It should be empty implementation.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to 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
[jira] [Commented] (FLINK-17925) Fix Filesystem options to default values and types
[
https://issues.apache.org/jira/browse/FLINK-17925?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17117398#comment-17117398
]
Jingsong Lee commented on FLINK-17925:
--
I have no obvious tendency, but I think "process-time" can be understood well.
> Fix Filesystem options to default values and types
> --
>
> Key: FLINK-17925
> URL: https://issues.apache.org/jira/browse/FLINK-17925
> Project: Flink
> Issue Type: Bug
> Components: Connectors / FileSystem
>Reporter: Jingsong Lee
>Assignee: Jingsong Lee
>Priority: Blocker
> Labels: pull-request-available
> Fix For: 1.11.0
>
>
> Fix Filesystem options:
> * Throws unsupported exception when using metastore commit policy for
> filesystem table, Filesystem connector has an empty implementation in
> {{TableMetaStoreFactory}}. We should avoid user configuring this policy.
> * Default value of "sink.partition-commit.trigger" should be "process-time".
> Users are hard to figure out what is wrong when they don't have watermark. We
> can set "sink.partition-commit.trigger" to "process-time" to have better
> out-of-box experience.
> * The type of "sink.rolling-policy.file-size" should be MemoryType.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on pull request #12334: [FLINK-17934][FLINK-17955][fs-connector] StreamingFileWriter should set chainingStrategy & BucketLifeCycleListener should just in Buc
flinkbot edited a comment on pull request #12334: URL: https://github.com/apache/flink/pull/12334#issuecomment-633879796 ## CI report: * 054aceb94feb7fd66916f4502de647b9b0f0a949 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2190) * e3f49ff243a1ef02b1fe3caeb4001e42d2c38636 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] flinkbot edited a comment on pull request #12275: [FLINK-16021][table-common] DescriptorProperties.putTableSchema does …
flinkbot edited a comment on pull request #12275: URL: https://github.com/apache/flink/pull/12275#issuecomment-631915106 ## CI report: * 2b848a7895db6b1b5263a5242e39b822fa30cecc Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2240) * e027ff0822c8621eb8f061888cc15fd7465133ba Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2241) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[jira] [Commented] (FLINK-17955) BucketLifeCycleListener should just in Buckets
[ https://issues.apache.org/jira/browse/FLINK-17955?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17117391#comment-17117391 ] Jingsong Lee commented on FLINK-17955: -- PR: [https://github.com/apache/flink/pull/12334] > BucketLifeCycleListener should just in Buckets > -- > > Key: FLINK-17955 > URL: https://issues.apache.org/jira/browse/FLINK-17955 > Project: Flink > Issue Type: Bug > Components: Connectors / FileSystem >Reporter: Jingsong Lee >Assignee: Jingsong Lee >Priority: Major > Fix For: 1.11.0 > > > We should keep BucketLifeCycleListener just in runtime. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-17955) BucketLifeCycleListener should just in Buckets
Jingsong Lee created FLINK-17955: Summary: BucketLifeCycleListener should just in Buckets Key: FLINK-17955 URL: https://issues.apache.org/jira/browse/FLINK-17955 Project: Flink Issue Type: Bug Components: Connectors / FileSystem Reporter: Jingsong Lee Assignee: Jingsong Lee Fix For: 1.11.0 We should keep BucketLifeCycleListener just in runtime. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (FLINK-17925) Fix Filesystem options to default values and types
[
https://issues.apache.org/jira/browse/FLINK-17925?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17117388#comment-17117388
]
Jark Wu commented on FLINK-17925:
-
I prefer "processing-time" because it is the concept glossary in Flink,
{{TimeCharacteristic#ProcessingTime}}.
> Fix Filesystem options to default values and types
> --
>
> Key: FLINK-17925
> URL: https://issues.apache.org/jira/browse/FLINK-17925
> Project: Flink
> Issue Type: Bug
> Components: Connectors / FileSystem
>Reporter: Jingsong Lee
>Assignee: Jingsong Lee
>Priority: Blocker
> Labels: pull-request-available
> Fix For: 1.11.0
>
>
> Fix Filesystem options:
> * Throws unsupported exception when using metastore commit policy for
> filesystem table, Filesystem connector has an empty implementation in
> {{TableMetaStoreFactory}}. We should avoid user configuring this policy.
> * Default value of "sink.partition-commit.trigger" should be "process-time".
> Users are hard to figure out what is wrong when they don't have watermark. We
> can set "sink.partition-commit.trigger" to "process-time" to have better
> out-of-box experience.
> * The type of "sink.rolling-policy.file-size" should be MemoryType.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on pull request #12275: [FLINK-16021][table-common] DescriptorProperties.putTableSchema does …
flinkbot edited a comment on pull request #12275: URL: https://github.com/apache/flink/pull/12275#issuecomment-631915106 ## CI report: * 2b848a7895db6b1b5263a5242e39b822fa30cecc Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2240) * e027ff0822c8621eb8f061888cc15fd7465133ba UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] wuchong commented on pull request #12320: [FLINK-17887][table][connector] Improve interface of ScanFormatFactory and SinkFormatFactory
wuchong commented on pull request #12320: URL: https://github.com/apache/flink/pull/12320#issuecomment-634421914 Hi @twalthr , I can see the benefit to have a hierarchy in the format interfaces. What do you think about to use the first proposal in our offline discussion? That `DecodingFormat#createRuntimeDecoder()` and `EncodingFormat#createRuntimeEncoder()`. I don't have a clear idea how to give a suitable method name for `SourceFormat#createXxxx`. This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] godfreyhe commented on pull request #11397: [FLINK-16217] [sql-client] catch SqlExecutionException for all callXX methods
godfreyhe commented on pull request #11397: URL: https://github.com/apache/flink/pull/11397#issuecomment-634418854 @twalthr I have updated this pr, please have a look if you have time, thanks This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[jira] [Commented] (FLINK-17953) OverWindow doesn't support to order by non-time attribute in batch mode for Table API program
[
https://issues.apache.org/jira/browse/FLINK-17953?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17117379#comment-17117379
]
Dian Fu commented on FLINK-17953:
-
In streaming jobs, there are also the same kind of problems if the source is
defined through DDL.
For the following job:
{code}
source_ddl = """
create table input(
id INT,
sales FLOAT,
word VARCHAR,
ts TIMESTAMP(3),
WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
) with (
'connector' = 'filesystem',
'format' = 'csv',
'path' = '{}'
)
""".format(input_data_path)
t_env.execute_sql(source_ddl)
sink_ddl = """
create table results(
id INT,
total_sales FLOAT
) with (
'connector' = 'filesystem',
'format' = 'csv',
'path' = '{}'
)
""".format(result_data_path)
t_env.execute_sql(sink_ddl)
table = t_env.from_path("input").over_window(
Over.partition_by("id").order_by("ts").preceding("2.rows").following("current_row").alias('w'))
\
.select("id, sum(sales) over w as total_sales")
{code}
It will also throw the same kind of exception as described in this JIRA.
> OverWindow doesn't support to order by non-time attribute in batch mode for
> Table API program
> -
>
> Key: FLINK-17953
> URL: https://issues.apache.org/jira/browse/FLINK-17953
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.9.0, 1.10.0, 1.11.0
>Reporter: Dian Fu
>Priority: Major
>
> For a simple batch job tested in blink planner:
> {code:java}
> INSERT INTO results
> SELECT id, sum(sales)
> OVER (PARTITION BY id ORDER BY ts ROWS BETWEEN 2 PRECEDING AND 0 FOLLOWING)
> FROM input
> {code}
> It could pass if written in SQL. However, if we rewrite it in Table API, it
> will throw the following exception:
> {code:java}
> py4j.protocol.Py4JJavaError: An error occurred while calling
> o85.select.py4j.protocol.Py4JJavaError: An error occurred while calling
> o85.select.: org.apache.flink.table.api.ValidationException: Ordering must be
> defined on a time attribute. at
> org.apache.flink.table.planner.expressions.PlannerTypeInferenceUtilImpl.validateArguments(PlannerTypeInferenceUtilImpl.java:112)
> at
> org.apache.flink.table.planner.expressions.PlannerTypeInferenceUtilImpl.runTypeInference(PlannerTypeInferenceUtilImpl.java:71)
> at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.runLegacyTypeInference(ResolveCallByArgumentsRule.java:218)
> at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.lambda$visit$2(ResolveCallByArgumentsRule.java:134)
> at java.util.Optional.orElseGet(Optional.java:267) at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.visit(ResolveCallByArgumentsRule.java:134)
> at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.visit(ResolveCallByArgumentsRule.java:89)
> at
> org.apache.flink.table.expressions.ApiExpressionVisitor.visit(ApiExpressionVisitor.java:39)
> at
> org.apache.flink.table.expressions.UnresolvedCallExpression.accept(UnresolvedCallExpression.java:132)
> at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.visit(ResolveCallByArgumentsRule.java:124)
> at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.visit(ResolveCallByArgumentsRule.java:89)
> at
> org.apache.flink.table.expressions.ApiExpressionVisitor.visit(ApiExpressionVisitor.java:39)
> at
> org.apache.flink.table.expressions.UnresolvedCallExpression.accept(UnresolvedCallExpression.java:132)
> at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule.lambda$apply$0(ResolveCallByArgumentsRule.java:83)
> at java.util.stream.ReferencePipeline$7$1.accept(ReferencePipeline.java:267)
> at
> java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1374)
> at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481) at
> java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
> at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708)
> at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) at
> java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:499) at
> org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule.apply(ResolveCallByArgumentsRule.java:84)
> at
> org.apache.flink.table.expressions.resolver.ExpressionResolver.lambda$null$1(ExpressionResolver.java:211)
> at java.util.function.Function.lambda$andThen$1(Function.java:88) at
>
[jira] [Created] (FLINK-17954) Do not multiplex remote function state into single PersistedTable
Tzu-Li (Gordon) Tai created FLINK-17954:
---
Summary: Do not multiplex remote function state into single
PersistedTable
Key: FLINK-17954
URL: https://issues.apache.org/jira/browse/FLINK-17954
Project: Flink
Issue Type: Task
Components: Stateful Functions
Affects Versions: statefun-2.0.1, statefun-2.1.0
Reporter: Tzu-Li (Gordon) Tai
We are currently multiplexing multiple remote function's user value states into
a single {{PersistedTable}}, using the state name as the table key.
This is not nice since:
- It does not allow individual states to have different properties, such as TTL
expiration.
- We are restricted to only value states for remote functions
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] zzchun opened a new pull request #12348: [hotfix] fix typo in ThreadDumpInfo
zzchun opened a new pull request #12348: URL: https://github.com/apache/flink/pull/12348 ## What is the purpose of the change fix typo in ThreadDumpInfo ## 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, Kubernetes/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 to 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
[GitHub] [flink] flinkbot edited a comment on pull request #12275: [FLINK-16021][table-common] DescriptorProperties.putTableSchema does …
flinkbot edited a comment on pull request #12275: URL: https://github.com/apache/flink/pull/12275#issuecomment-631915106 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink-web] klion26 opened a new pull request #342: [FLINK-17926] Fix the build problem of docker image
klion26 opened a new pull request #342: URL: https://github.com/apache/flink-web/pull/342 Currently, we use Ubuntu:18.10 to build the docker image, but Ubuntu:18.10 has reached the end of life. In this commit we change to the closest LTS version 18.04 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] kl0u commented on a change in pull request #12179: [FLINK-16144] get client.timeout for the client, with a fallback to the akka.client…
kl0u commented on a change in pull request #12179:
URL: https://github.com/apache/flink/pull/12179#discussion_r429985424
##
File path:
flink-clients/src/main/java/org/apache/flink/client/cli/ClientOptions.java
##
@@ -0,0 +1,68 @@
+/*
+ * 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.client.cli;
+
+import org.apache.flink.configuration.AkkaOptions;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.TimeUtils;
+
+import java.time.Duration;
+import java.util.Optional;
+
+/**
+ * Describes a client configuration parameter.
+ */
+public class ClientOptions {
+
Review comment:
I think that the `CLIENT_TIMEOUT` should subsume the
`EMBEDDED_RPC_TIMEOUT` and the `EMBEDDED_RPC_RETRY_PERIOD` should become
`CLIENT_RETRY_PERIOD`. So the class should have these 2 options:
```
public static final ConfigOption CLIENT_TIMEOUT =
ConfigOptions.key("client.timeout")
.durationType()
.defaultValue(Duration.ofMillis(60 *
1000))
.withDescription("The client timeout
(in ms) when executing commands from Flink's CLI or its " +
"clients (default
1min).");
public static final ConfigOption CLIENT_RETRY_PERIOD =
ConfigOptions.key("client.retry-period")
.durationType()
.defaultValue(Duration.ofMillis(2000))
.withDescription("The interval (in ms)
between consecutive retries of failed attempts to execute " +
"commands through the
CLI or Flink's clients, wherever retry is supported (default 2sec).");
```
##
File path:
flink-clients/src/main/java/org/apache/flink/client/cli/ClientOptions.java
##
@@ -0,0 +1,68 @@
+/*
+ * 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.client.cli;
+
+import org.apache.flink.configuration.AkkaOptions;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.TimeUtils;
+
+import java.time.Duration;
+import java.util.Optional;
+
+/**
+ * Describes a client configuration parameter.
+ */
+public class ClientOptions {
+
Review comment:
You can simply remove the `EMBEDDED_RPC_TIMEOUT` because this was
introduced in this release, so it is not released yet.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to 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
[GitHub] [flink] flinkbot edited a comment on pull request #12331: [FLINK-17936][table] Introduce new type inference for AS
flinkbot edited a comment on pull request #12331: URL: https://github.com/apache/flink/pull/12331#issuecomment-633846311 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] flinkbot commented on pull request #12333: [FLINK-17937][hive] Change some hive connector tests to IT cases
flinkbot commented on pull request #12333: URL: https://github.com/apache/flink/pull/12333#issuecomment-633861568 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] danny0405 commented on a change in pull request #12275: [FLINK-16021][table-common] DescriptorProperties.putTableSchema does …
danny0405 commented on a change in pull request #12275:
URL: https://github.com/apache/flink/pull/12275#discussion_r430239805
##
File path:
flink-table/flink-table-common/src/main/java/org/apache/flink/table/descriptors/DescriptorProperties.java
##
@@ -610,7 +626,9 @@ public DataType getDataType(String key) {
public Optional getOptionalTableSchema(String key) {
// filter for number of fields
final int fieldCount = properties.keySet().stream()
- .filter((k) -> k.startsWith(key) && k.endsWith('.' +
TABLE_SCHEMA_NAME))
+ .filter((k) -> k.startsWith(key)
+ // "key." is the prefix.
+ &&
SCHEMA_COLUMN_NAME_SUFFIX.matcher(k.substring(key.length() + 1)).matches())
Review comment:
I think the regex matching is a more general way to match column names
for the long run, just remove some specific keys seems hacky from my side.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to 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
[GitHub] [flink] azagrebin commented on a change in pull request #12278: [FLINK-17019][runtime] Fulfill slot requests in request order
azagrebin commented on a change in pull request #12278: URL: https://github.com/apache/flink/pull/12278#discussion_r430441131 ## File path: flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolImpl.java ## @@ -112,6 +114,9 @@ /** The requests that are waiting for the resource manager to be connected. */ private final LinkedHashMap waitingForResourceManager; + /** Maps a request to its allocation. */ + private final BiMap requestedAllocations; Review comment: Looking into the implementation of `DualKeyLinkedMap` for `waitingForResourceManager`, it seems we can just remove the first matching `SlotRequestId` and then remap the orphaned `SlotRequestId` to its `AllocationID`. The original insertion ordering should not suffer in `DualKeyLinkedMap.aMap`. If so, we could remove `requestedAllocations`. ## File path: flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolImpl.java ## @@ -648,26 +648,8 @@ boolean offerSlot( slotOffer.getResourceProfile(), taskManagerGateway); - // check whether we have request waiting for this slot - PendingRequest pendingRequest = pendingRequests.removeKeyB(allocationID); Review comment: I am not sure about all consequences of this change for the existing scheduling. I mean that we do not respect SlotRequestId->AllocationID by accepting the slot offer. Would it make sense to keep this behaviour configurable for now depending on scheduling strategy? Or this complication is not needed? This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] flinkbot commented on pull request #12326: [FLINK-16577] [table-planner-blink] Fix numeric type mismatch error in column interval relmetadata
flinkbot commented on pull request #12326: URL: https://github.com/apache/flink/pull/12326#issuecomment-633780190 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] wtog commented on a change in pull request #12179: [FLINK-16144] get client.timeout for the client, with a fallback to the akka.client…
wtog commented on a change in pull request #12179:
URL: https://github.com/apache/flink/pull/12179#discussion_r430522062
##
File path:
flink-clients/src/main/java/org/apache/flink/client/cli/ClientOptions.java
##
@@ -0,0 +1,68 @@
+/*
+ * 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.client.cli;
+
+import org.apache.flink.configuration.AkkaOptions;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.TimeUtils;
+
+import java.time.Duration;
+import java.util.Optional;
+
+/**
+ * Describes a client configuration parameter.
+ */
+public class ClientOptions {
+
Review comment:
hi @kl0u, need i write like this ?
get client timeout value
from configurated client_timeout value orelse
execution.embedded-rpc-timeout value orelse
AkkaOptions.CLIENT_TIMEOUT orelse
client_timeout default value?
```java
public static Duration getClientTimeout(Configuration configuration) {
Optional timeoutOptional =
configuration.getOptional(CLIENT_TIMEOUT);
if (timeoutOptional.isPresent()) {
return timeoutOptional.get();
} else {
Optional rpcClientTimeout =
configuration.getOptional(EMBEDDED_RPC_TIMEOUT);
if (rpcClientTimeout.isPresent()) {
return rpcClientTimeout.get();
} else {
Optional akkaClientTimeout =
configuration.getOptional(AkkaOptions.CLIENT_TIMEOUT);
if (akkaClientTimeout.isPresent()) {
return
TimeUtils.parseDuration(akkaClientTimeout.get());
} else {
return CLIENT_TIMEOUT.defaultValue();
}
}
}
}
```
##
File path:
flink-clients/src/main/java/org/apache/flink/client/cli/ClientOptions.java
##
@@ -0,0 +1,68 @@
+/*
+ * 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.client.cli;
+
+import org.apache.flink.configuration.AkkaOptions;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.TimeUtils;
+
+import java.time.Duration;
+import java.util.Optional;
+
+/**
+ * Describes a client configuration parameter.
+ */
+public class ClientOptions {
+
Review comment:
hi @kl0u, need i write like this ?
get client timeout value
from configurated **CLIENT_TIMEOUT** value orelse
**EMBEDDED_RPC_TIMEOUT** value orelse
**AkkaOptions.CLIENT_TIMEOUT** orelse
**CLIENT_TIMEOUT** default value?
```java
public static Duration getClientTimeout(Configuration configuration) {
Optional timeoutOptional =
configuration.getOptional(CLIENT_TIMEOUT);
if (timeoutOptional.isPresent()) {
return timeoutOptional.get();
} else {
Optional rpcClientTimeout =
configuration.getOptional(EMBEDDED_RPC_TIMEOUT);
if (rpcClientTimeout.isPresent()) {
[GitHub] [flink] flinkbot edited a comment on pull request #12325: [FLINK-17931] Document fromValues clause in Table API
flinkbot edited a comment on pull request #12325: URL: https://github.com/apache/flink/pull/12325#issuecomment-633600802 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] azagrebin commented on a change in pull request #12256: [FLINK-17018][runtime] Allocates slots in bulks for pipelined region scheduling
azagrebin commented on a change in pull request #12256:
URL: https://github.com/apache/flink/pull/12256#discussion_r429801548
##
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotOccupationTest.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.runtime.jobmaster.slotpool;
+
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import
org.apache.flink.runtime.executiongraph.utils.SimpleAckingTaskManagerGateway;
+import org.apache.flink.runtime.jobmanager.scheduler.Locality;
+import org.apache.flink.runtime.jobmanager.slots.TestingSlotOwner;
+import org.apache.flink.runtime.jobmaster.LogicalSlot;
+import org.apache.flink.runtime.jobmaster.SlotRequestId;
+import org.apache.flink.runtime.taskmanager.LocalTaskManagerLocation;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests whether the slot occupation works correctly.
+ */
+public class SlotOccupationTest extends TestLogger {
+
+ @Test
+ public void testSingleTaskOccupyingSlotIndefinitely() {
+ final PhysicalSlot physicalSlot = createPhysicalSlot();
+ allocateSingleLogicalSlotFromPhysicalSlot(physicalSlot, true);
+
+ assertTrue(physicalSlot.willBeOccupiedIndefinitely());
+ }
+
+ @Test
+ public void testSingleTaskNotOccupyingSlotIndefinitely() {
+ final PhysicalSlot physicalSlot = createPhysicalSlot();
+ allocateSingleLogicalSlotFromPhysicalSlot(physicalSlot, true);
+
+ assertTrue(physicalSlot.willBeOccupiedIndefinitely());
Review comment:
```suggestion
allocateSingleLogicalSlotFromPhysicalSlot(physicalSlot, false);
assertFalse(physicalSlot.willBeOccupiedIndefinitely());
```
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolImpl.java
##
@@ -469,6 +469,47 @@ public void releaseSlot(@Nonnull SlotRequestId
slotRequestId, @Nullable Throwabl
.collect(Collectors.toList());
}
+ @Override
+ public boolean isSlotRequestBulkFulfillable(final
PhysicalSlotRequestBulk slotRequestBulk) {
+ final Set assignedSlots = new
HashSet<>(slotRequestBulk.getFulfilledRequests().values());
+ final Set reusableSlots =
getReusableSlots(assignedSlots);
+ return
areRequestsFulfillableWithSlots(slotRequestBulk.getPendingRequests().values(),
reusableSlots);
+ }
+
+ private Set getReusableSlots(final Set
slotsToExclude) {
+ return Stream
+ .concat(
+ getAvailableSlotsInformation().stream(),
+ getAllocatedSlotsInformation().stream())
+ .filter(slotInfo ->
!slotInfo.willBeOccupiedIndefinitely())
+ .filter(slotInfo ->
!slotsToExclude.contains(slotInfo.getAllocationId()))
+ .collect(Collectors.toSet());
+ }
+
+ private static boolean areRequestsFulfillableWithSlots(
+ final Collection requests,
+ final Set slots) {
+
+ final Set remainingSlots = new HashSet<>(slots);
+ for (PhysicalSlotRequest request : requests) {
+ final Optional matchedSlot =
findMatchingSlotForRequest(request, remainingSlots);
+ if (matchedSlot.isPresent()) {
+ remainingSlots.remove(matchedSlot.get());
+ } else {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static Optional findMatchingSlotForRequest(
+ final PhysicalSlotRequest request,
+ final Collection slots) {
+
+ final ResourceProfile requiredResource =
request.getSlotProfile().getPhysicalSlotResourceProfile();
[GitHub] [flink] wuchong merged pull request #12254: [FLINK-17802][kafka] Set offset commit only if group id is configured for new Kafka Table source
wuchong merged pull request #12254: URL: https://github.com/apache/flink/pull/12254 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] tillrohrmann commented on a change in pull request #12315: [FLINK-17917][yarn] Ignore the external resource with a value of 0 in…
tillrohrmann commented on a change in pull request #12315:
URL: https://github.com/apache/flink/pull/12315#discussion_r429966710
##
File path:
flink-yarn/src/main/java/org/apache/flink/yarn/ResourceInformationReflector.java
##
@@ -151,7 +151,9 @@ void setResourceInformationUnSafe(Object resource, String
resourceName, long amo
for (int i = 2; i < externalResourcesInfo.length; i++) {
final String name = (String)
resourceInformationGetNameMethod.invoke(externalResourcesInfo[i]);
final long value = (long)
resourceInformationGetValueMethod.invoke(externalResourcesInfo[i]);
- externalResources.put(name, value);
+ if (value > 0) {
+ externalResources.put(name, value);
+ }
Review comment:
I think the proper solution is to enforce this invariant in the
`InternalContainerResource` constructor (filtering out entries with a value of
`0` or to adapt the `equals` method to not check entries with a value of `0`).
Otherwise we might run into the situation that sometime in the future when the
user can specify the GPU constraints on an operator and he sets it to `0` that
the resources won't match.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to 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
[GitHub] [flink] zhengcanbin opened a new pull request #12329: Release 1.11
zhengcanbin opened a new pull request #12329: URL: https://github.com/apache/flink/pull/12329 ## What is the purpose of the change *(For example: This pull request makes task deployment go through the blob server, rather than through RPC. That way we avoid re-transferring them on each deployment (during recovery).)* ## Brief change log *(for example:)* - *The TaskInfo is stored in the blob store on job creation time as a persistent artifact* - *Deployments RPC transmits only the blob storage reference* - *TaskManagers retrieve the TaskInfo from the blob cache* ## Verifying this change *(Please pick either of the following options)* This change is a trivial rework / code cleanup without any test coverage. *(or)* This change is already covered by existing tests, such as *(please describe tests)*. *(or)* This change added tests and can be verified as follows: *(example:)* - *Added integration tests for end-to-end deployment with large payloads (100MB)* - *Extended integration test for recovery after master (JobManager) failure* - *Added test that validates that TaskInfo is transferred only once across recoveries* - *Manually verified the change by running a 4 node cluser with 2 JobManagers and 4 TaskManagers, a stateful streaming program, and killing one JobManager and two TaskManagers during the execution, verifying that recovery happens correctly.* ## 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, Kubernetes/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 to 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
[GitHub] [flink] tillrohrmann closed pull request #12315: [FLINK-17917][yarn] Ignore the external resource with a value of 0 in…
tillrohrmann closed pull request #12315: URL: https://github.com/apache/flink/pull/12315 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] flinkbot commented on pull request #12331: [FLINK-17936][table] Introduce new type inference for AS
flinkbot commented on pull request #12331: URL: https://github.com/apache/flink/pull/12331#issuecomment-633836671 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[jira] [Updated] (FLINK-17926) Can't build flink-web docker image because of EOL of Ubuntu:18.10
[
https://issues.apache.org/jira/browse/FLINK-17926?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-17926:
---
Labels: pull-request-available (was: )
> Can't build flink-web docker image because of EOL of Ubuntu:18.10
> -
>
> Key: FLINK-17926
> URL: https://issues.apache.org/jira/browse/FLINK-17926
> Project: Flink
> Issue Type: Bug
> Components: Project Website
>Reporter: Congxian Qiu(klion26)
>Assignee: Congxian Qiu(klion26)
>Priority: Major
> Labels: pull-request-available
>
> Currently, the Dockerfile[1] in flink-web project is broken because of the
> EOL of Ubuntu 18.10[2], will encounter the error such as bellow when
> executing {{./run.sh}}
> {code:java}
> Err:3 http://security.ubuntu.com/ubuntu cosmic-security Release
> 404 Not Found [IP: 91.189.88.152 80]
> Ign:4 http://archive.ubuntu.com/ubuntu cosmic-updates InRelease
> Ign:5 http://archive.ubuntu.com/ubuntu cosmic-backports InRelease
> Err:6 http://archive.ubuntu.com/ubuntu cosmic Release
> 404 Not Found [IP: 91.189.88.142 80]
> Err:7 http://archive.ubuntu.com/ubuntu cosmic-updates Release
> 404 Not Found [IP: 91.189.88.142 80]
> Err:8 http://archive.ubuntu.com/ubuntu cosmic-backports Release
> 404 Not Found [IP: 91.189.88.142 80]
> Reading package lists...
> {code}
> The current LTS versions can be found in release website[2].
> Apache Flink docker image uses fedora:28[3], so it unaffected.
> As fedora does not have LTS release[4], I proposal to use Ubuntu for website
> here, and change the version from {{18.10}} to the closest LTS version
> {{18.04, tried locally, it works successfully.}}
> [1]
> [https://github.com/apache/flink-web/blob/bc66f0f0f463ab62a22e81df7d7efd301b76a6b4/docker/Dockerfile#L17]
> [2] [https://wiki.ubuntu.com/Releases]
>
> [3]https://github.com/apache/flink/blob/e92b2bf19bdf03ad3bae906dc5fa3781aeddb3ee/docs/docker/Dockerfile#L17
> [4]
> https://fedoraproject.org/wiki/Fedora_Release_Life_Cycle#Maintenance_Schedule
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] twalthr closed pull request #12331: [FLINK-17936][table] Introduce new type inference for AS
twalthr closed pull request #12331: URL: https://github.com/apache/flink/pull/12331 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] flinkbot commented on pull request #12348: [hotfix] fix typo in ThreadDumpInfo
flinkbot commented on pull request #12348: URL: https://github.com/apache/flink/pull/12348#issuecomment-634390040 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink-web] klion26 commented on pull request #342: [FLINK-17926] Fix the build problem of docker image
klion26 commented on pull request #342: URL: https://github.com/apache/flink-web/pull/342#issuecomment-633880466 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] flinkbot commented on pull request #12344: [FLINK-16694][ci] Limit number of dumped log lines
flinkbot commented on pull request #12344: URL: https://github.com/apache/flink/pull/12344#issuecomment-634010415 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] TisonKun merged pull request #12277: [FLINK-17230] Fix incorrect returned address of Endpoint for external Service of ClusterIP type
TisonKun merged pull request #12277: URL: https://github.com/apache/flink/pull/12277 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] flinkbot edited a comment on pull request #12321: Add document for writing Avro files with StreamingFileSink
flinkbot edited a comment on pull request #12321: URL: https://github.com/apache/flink/pull/12321#issuecomment-633558255 ## CI report: * 56bdc3a61f65cb30b48cf7f932520be02ebed734 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2143) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink-web] zjffdu commented on pull request #339: [blog] flink on zeppelin
zjffdu commented on pull request #339: URL: https://github.com/apache/flink-web/pull/339#issuecomment-633606084 Thanks @morsapaes , I have updated the PR This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] DashShen commented on a change in pull request #9898: [FLINK-14359][hbase] Create a module called flink-sql-connector-hbase…
DashShen commented on a change in pull request #9898:
URL: https://github.com/apache/flink/pull/9898#discussion_r430186202
##
File path: flink-connectors/flink-sql-connector-hbase/pom.xml
##
@@ -0,0 +1,102 @@
+
+
+http://maven.apache.org/POM/4.0.0;
+xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance;
+xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd;>
+ 4.0.0
+
+ flink-connectors
+ org.apache.flink
+ 1.10-SNAPSHOT
+
+
+
flink-sql-connector-hbase_${scala.binary.version}
+ flink-sql-connector-hbase
+
+ jar
+
+
+
+ org.apache.flink
+
flink-hbase_${scala.binary.version}
Review comment:
the latest hbase connector has already rename to flink-connector-hbase
This is an automated message from the Apache Git Service.
To respond to the message, please log on to 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
[GitHub] [flink] sjwiesman commented on a change in pull request #12257: [FLINK-17076][docs] Revamp Kafka Connector Documentation
sjwiesman commented on a change in pull request #12257: URL: https://github.com/apache/flink/pull/12257#discussion_r430690241 ## File path: docs/dev/connectors/kafka.md ## @@ -152,16 +86,14 @@ The constructor accepts the following arguments: - "bootstrap.servers" (comma separated list of Kafka brokers) - "group.id" the id of the consumer group Review comment: Will update This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] flinkbot edited a comment on pull request #12028: [FLINK-17553][table]fix plan error when constant exists in group window key
flinkbot edited a comment on pull request #12028: URL: https://github.com/apache/flink/pull/12028#issuecomment-625656822 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] lgo opened a new pull request #12345: [FLINK-17288] [Runtime/StateBackends] Add RocksDB SST ingestion for batch writes
lgo opened a new pull request #12345: URL: https://github.com/apache/flink/pull/12345 ## WIP I was hoping to get a first round of feedback for this implementation. This branch is currently passing tests, but there was additional work to clean it up: - [ ] More testing of the new implementation. - [ ] Add benchmarking to RocksDB for the two implementations used here. - [ ] Add more configuration parameters to pass throguh to the writers. - [ ] Compare performance of save-point recovery on our production instanace, as it has a large state. ## What is the purpose of the change This adds a new mode of batch writing keys into RocksDB, via `RocksDBSSTIngestWriter`, which should provide a considerable performance improvement to some operations such as save-point recovery. This is in reference to the discussion on the users maillist I brought up, [here](http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/RocksDB-savepoint-recovery-performance-improvements-td35238.html), as well as the previously reported ticket: https://issues.apache.org/jira/browse/FLINK-17288. The first commit is also replaces one use of RocksDB with a more efficient operation (`deleteRange`). ## Brief change log (commit 1) - Replaced an iterate and `delete` operation with `RocksDB#deleteRange`. (commit 2) - Refactored the use of `RocksDBWriteBatchWrapper` into using a factory (`RocksDBWriterFactory`) and interface (`RocksDBWriter`), in preperation of adding a second implementation. (commit 3) - Added `RockSDBSSTWriter`, which is a basic wrapper for `SstFileWriter` in order to create `sst` files. - Added `RocksDBSSTIngestWriter`, which uses the `RockSDBSSTWriter`, and provides a write-interface for batch writing k/v into RocksDB. This includes flushing and handling multiple column-families. - Added new configuration for opting into the writer, as well as tuning parameters. This configuration was plumbed into `RocksDBWriterFactory`. ## Verifying this change This change is already covered by existing tests, such as: - RocksDB savepoint and checkpoint tests. This change added tests and can be verified as follows: - Added new tests for the `RocksDBSSTWriter` and `RocksDBSSTIngestWriter`. - [ ] **TODO** Add more rigourous tests for the new implementation. - [ ] **TODO** Extend existing tests to test both writer implementations. - [ ] **TODO** Manually verified the change running on a cluster. - [ ] **TODO** Write benchmarks for https://github.com/facebook/rocksdb to compare the two writing methods. ## 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): **yes** - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: **yes** - The S3 file system connector: **no** ## Documentation - Does this pull request introduce a new feature? **yes** - If yes, how is the feature documented? **TODO: not yet documented** This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] LakeShen commented on pull request #11499: [FLINK-16681][Connectors/JDBC]Jdbc JDBCOutputFormat and JDBCLookupFunction PreparedStatement loss connection, if long time not records to wr
LakeShen commented on pull request #11499: URL: https://github.com/apache/flink/pull/11499#issuecomment-633992991 Sorry，I've been concentrating on my work lately，I don't have free time to solve this problem. If you hava free time to solve it , please to do that. I'm so sorry to say it. This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] lirui-apache commented on a change in pull request #12323: [FLINK-17925][fs-connector] Fix Filesystem options to default values and types
lirui-apache commented on a change in pull request #12323:
URL: https://github.com/apache/flink/pull/12323#discussion_r430220278
##
File path:
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/PartitionCommitPolicy.java
##
@@ -121,4 +122,20 @@
}
}).collect(Collectors.toList());
}
+
+ /**
+* Validate commit policy.
+*/
+ static void validatePolicyChain(boolean isEmptyMetastore, String
policyKind) {
+ if (policyKind != null) {
+ String[] policyStrings = policyKind.split(",");
+ for (String policy : policyStrings) {
+ if (isEmptyMetastore &&
METASTORE.equalsIgnoreCase(policy)) {
+ throw new ValidationException(
+ "Can not configure a
metastore partition commit policy for" +
+ " a
table without metastore.");
Review comment:
I'm not sure if the error message is easy to understand. What is "a
table without metastore"?
##
File path:
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemOptions.java
##
@@ -132,12 +133,12 @@
public static final ConfigOption SINK_PARTITION_COMMIT_TRIGGER =
key("sink.partition-commit.trigger")
.stringType()
- .defaultValue("partition-time")
+ .defaultValue("process-time")
.withDescription("Trigger type for
partition commit:" +
Review comment:
I find this config a little bit weird.
`process-time` means "**process-time** > partition dir create time + delay"
`partition-time` means "watermark > **partition-time** + delay"
Seems the two options configure different parts of the inequation. What if
user wants to compare watermark with partition dir create time, or compare
process-time with time extracted from partition values?
##
File path:
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/EmptyMetaStoreFactory.java
##
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.filesystem;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.core.fs.Path;
+
+import java.util.LinkedHashMap;
+import java.util.Optional;
+
+/**
+ * Empty implementation {@link TableMetaStoreFactory}.
Review comment:
It's not really an **empty** implementation right? Because we still need
it to get the location path?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to 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
[GitHub] [flink] dawidwys opened a new pull request #12325: [FLINK-17931] Document fromValues clause in Table API
dawidwys opened a new pull request #12325: URL: https://github.com/apache/flink/pull/12325 ## What is the purpose of the change Add documentation on the `fromValues` clause. ## 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, Kubernetes/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 to 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
[GitHub] [flink] zjuwangg commented on pull request #12028: [FLINK-17553][table]fix plan error when constant exists in group window key
zjuwangg commented on pull request #12028: URL: https://github.com/apache/flink/pull/12028#issuecomment-633774808 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] flinkbot edited a comment on pull request #12344: [FLINK-16694][ci] Limit number of dumped log lines
flinkbot edited a comment on pull request #12344: URL: https://github.com/apache/flink/pull/12344#issuecomment-634021233 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] flinkbot commented on pull request #12338: [hotfix][tests] Document jepsen correctness model
flinkbot commented on pull request #12338: URL: https://github.com/apache/flink/pull/12338#issuecomment-633933390 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] flinkbot edited a comment on pull request #12345: [FLINK-17288] [Runtime/StateBackends] Add RocksDB SST ingestion for batch writes
flinkbot edited a comment on pull request #12345: URL: https://github.com/apache/flink/pull/12345#issuecomment-634034354 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] nielsbasjes commented on a change in pull request #11245: [FLINK-15794][Kubernetes] Generate the Kubernetes default image version
nielsbasjes commented on a change in pull request #11245:
URL: https://github.com/apache/flink/pull/11245#discussion_r430425933
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/util/EnvironmentInformation.java
##
@@ -103,6 +103,15 @@ public static String getGitCommitTimeString() {
return getVersionsInstance().gitCommitTimeStr;
}
+ /**
+* Returns the exact name of the Dockerimage for Flink that is needed
to run in.
+*
+* @return The "name:tag" of the Flink docker image.
+*/
+ public static String getFlinkDockerImageNameAndTag() {
Review comment:
I expect that running from the docker image will be a major factor in
running Flink in the next few years. Because I expect this to go beyond
Kubernetes I put this in the current central place.
##
File path: docs/ops/deployment/kubernetes.md
##
@@ -159,6 +159,8 @@ with the `kubectl` command:
## Appendix
### Common cluster resource definitions
+The Deployment definitions use the pre-built image `flink:{% if site.is_stable
%}{{site.version}}-scala{{site.scala_version_suffix}}`{% else %}latest` (which
contains the latest released Flink built against Scala 2.12){% endif %} which
can be found [on Docker Hub](https://hub.docker.com/_/flink/).
+The image is built from this [Github
repository](https://github.com/apache/flink-docker).
Review comment:
Yes, I fixed this. I've added a bit in the central docker documentation
files instead.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to 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
[GitHub] [flink] wuchong merged pull request #12343: [FLINK-17939][docs-zh] Translate "Python Table API Installation" page into Chinese
wuchong merged pull request #12343: URL: https://github.com/apache/flink/pull/12343 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] XBaith commented on a change in pull request #12237: [FLINK-17290] [chinese-translation, Documentation / Training] Transla…
XBaith commented on a change in pull request #12237:
URL: https://github.com/apache/flink/pull/12237#discussion_r430014031
##
File path: docs/training/streaming_analytics.zh.md
##
@@ -27,125 +27,101 @@ under the License.
* This will be replaced by the TOC
{:toc}
-## Event Time and Watermarks
+## 事件时间和水印
-### Introduction
+### 简介
-Flink explicitly supports three different notions of time:
+Flink 明确的支持以下三种事件时间:
-* _event time:_ the time when an event occurred, as recorded by the device
producing (or storing) the event
+* _事件时间:_ 事件产生的时间，记录的是设备生产(或者存储)事件的时间
-* _ingestion time:_ a timestamp recorded by Flink at the moment it ingests the
event
+* _摄取时间:_ Flink 提取事件时记录的时间戳
-* _processing time:_ the time when a specific operator in your pipeline is
processing the event
+* _处理时间:_ Flink 中通过特定的操作处理事件的时间
-For reproducible results, e.g., when computing the maximum price a stock
reached during the first
-hour of trading on a given day, you should use event time. In this way the
result won't depend on
-when the calculation is performed. This kind of real-time application is
sometimes performed using
-processing time, but then the results are determined by the events that happen
to be processed
-during that hour, rather than the events that occurred then. Computing
analytics based on processing
-time causes inconsistencies, and makes it difficult to re-analyze historic
data or test new
-implementations.
+为了获得可重现的结果，例如在计算过去的特定一天里第一个小时股票的最高价格时，我们应该使用事件时间。这样的话，无论
+什么时间去计算都不会影响输出结果。然而有些人，在实时计算应用时使用处理时间，这样的话，输出结果就会被处理时间点所决
+定，而不是事件的生成时间。基于处理时间会导致多次计算的结果不一致，也可能会导致重新分析历史数据和测试变得异常困难。
-### Working with Event Time
+### 使用事件时间
-By default, Flink will use processing time. To change this, you can set the
Time Characteristic:
+Flink 在默认情况下使用处理时间。也可以通过如下配置来告诉 Flink 选择哪种事件时间:
{% highlight java %}
final StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
{% endhighlight %}
-If you want to use event time, you will also need to supply a Timestamp
Extractor and Watermark
-Generator that Flink will use to track the progress of event time. This will
be covered in the
-section below on [Working with Watermarks]({% link
-training/streaming_analytics.zh.md %}#working-with-watermarks), but first we
should explain what
-watermarks are.
+如果想要使用事件时间，则需要额外给 Flink 提供一个时间戳的提取器和水印，Flink 将使用它们来跟踪事件时间的进度。这
+将在选节[使用水印]({% linktutorials/streaming_analytics.md %}#使用水印)中介绍，但是首先我们需要解释一下
+水印是什么。
-### Watermarks
+### 水印
-Let's work through a simple example that will show why watermarks are needed,
and how they work.
+让我们通过一个简单的示例来演示，该示例将说明为什么需要水印及其工作方式。
-In this example you have a stream of timestamped events that arrive somewhat
out of order, as shown
-below. The numbers shown are timestamps that indicate when these events
actually occurred. The first
-event to arrive happened at time 4, and it is followed by an event that
happened earlier, at time 2,
-and so on:
+在此示例中，我们将看到带有混乱时间戳的事件流，如下所示。显示的数字表达的是这些事件实际发生时间的时间戳。到达的
+第一个事件发生在时间4，随后发生的事件发生在更早的时间2，依此类推：
··· 23 19 22 24 21 14 17 13 12 15 9 11 7 2 4 →
-Now imagine that you are trying create a stream sorter. This is meant to be an
application that
-processes each event from a stream as it arrives, and emits a new stream
containing the same events,
-but ordered by their timestamps.
+假设我们要对数据流排序，我们想要达到的目的是：应用程序应该在数据流里的事件到达时就处理每个事件，并发出包含相同
+事件但按其时间戳排序的新流。
-Some observations:
+让我们重新审视这些数据:
-(1) The first element your stream sorter sees is the 4, but you can't just
immediately release it as
-the first element of the sorted stream. It may have arrived out of order, and
an earlier event might
-yet arrive. In fact, you have the benefit of some god-like knowledge of this
stream's future, and
-you can see that your stream sorter should wait at least until the 2 arrives
before producing any
-results.
+(1) 我们的排序器第一个看到的数据是4，但是我们不能立即将其作为已排序流的第一个元素释放。因为我们并不能确定它是
+有序的，并且较早的事件有可能并未到达。事实上，如果站在上帝视角，我们知道，必须要等到2到来时，排序器才可以有事件输出。
-*Some buffering, and some delay, is necessary.*
+*需要一些缓冲，需要一些时间，但这都是值得的*
-(2) If you do this wrong, you could end up waiting forever. First the sorter
saw an event from time
-4, and then an event from time 2. Will an event with a timestamp less than 2
ever arrive? Maybe.
-Maybe not. You could wait forever and never see a 1.
+(2) 接下来的这一步，如果我们选择的是固执的等待，我们永远不会有结果。首先，我们从时间4看到了一个事件，然后从时
+间2看到了一个事件。可是，时间戳小于2的事件接下来会不会到来呢？可能会，也可能不会。再次站在上帝视角，我们知道，我
+们永远不会看到1。
-*Eventually you have to be courageous and emit the 2 as the start of the
sorted stream.*
+*最终，我们必须勇于承担责任，并发出指令，把2作为已排序的事件流的开始*
-(3) What you need then is some sort of policy that defines when, for any given
timestamped event, to
-stop waiting for the arrival of earlier events.
+(3)然后，我们需要一种策略，该策略定义：对于任何给定时间戳的事件，Flink何时停止等待较早事件的到来。
-*This is precisely what watermarks do* — they define when to stop waiting for
earlier events.
+*这正是水印的作用* — 它们定义何时停止等待较早的事件。
[GitHub] [flink] flinkbot edited a comment on pull request #12281: [FLINK-17456][hive][test] Update hive connector tests to execute DDL …
flinkbot edited a comment on pull request #12281: URL: https://github.com/apache/flink/pull/12281#issuecomment-632080083 ## CI report: * 082d175c44b62e1848b2896f8478eb2091b39e86 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2135) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] flinkbot edited a comment on pull request #12251: [FLINK-17812][dist] Bundle reporters in plugins/ directory
flinkbot edited a comment on pull request #12251: URL: https://github.com/apache/flink/pull/12251#issuecomment-630873924 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] flinkbot edited a comment on pull request #12348: [hotfix] fix typo in ThreadDumpInfo
flinkbot edited a comment on pull request #12348: URL: https://github.com/apache/flink/pull/12348#issuecomment-634392378 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] austince commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor
austince commented on pull request #12056: URL: https://github.com/apache/flink/pull/12056#issuecomment-633650241 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] godfreyhe opened a new pull request #12326: [FLINK-16577] [table-planner-blink] Fix numeric type mismatch error in column interval relmetadata
godfreyhe opened a new pull request #12326: URL: https://github.com/apache/flink/pull/12326 ## What is the purpose of the change *ValueInterval requires the upper class should be consistent with lower class, while in some case (e.g. SELECT c, SUM(a) FROM T WHERE a > 0.1 AND a < 1 GROUP BY c), there types are different, and will trigger validation exception. This pr aims to fix the bug, the solution is converting numeric type to BigDecimal type.* ## Brief change log - *code cleanup for FlinkRelOptUtil* - *Convert numeric type to BigDecimal in column interval metadata handler* ## Verifying this change This change added tests and can be verified as follows: - *Updated existing metadata handler tests for validating the change* - *Added a plan test in AggregateTest for validating the example in the issue* ## 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, Kubernetes/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 to 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
[GitHub] [flink] HuangXingBo commented on pull request #12328: [FLINK-17722][python][build system] (followups) Keeps all jars of lib directory in CachedFiles
HuangXingBo commented on pull request #12328: URL: https://github.com/apache/flink/pull/12328#issuecomment-633799246 It is the nightly build in my free Azure account. https://dev.azure.com/hxbks2ks/FLINK-TEST/_build/results?buildId=282=artifacts=publishedArtifacts. And I test the wheel packages is the same as the built from the source code. This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] flinkbot edited a comment on pull request #12324: [hotfix] Remove generic row for HiveTableFactory
flinkbot edited a comment on pull request #12324: URL: https://github.com/apache/flink/pull/12324#issuecomment-633592835 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] zhengcanbin closed pull request #12327: [FLINK-17565][k8s] Backport to release-1.11
zhengcanbin closed pull request #12327: URL: https://github.com/apache/flink/pull/12327 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] flinkbot edited a comment on pull request #12313: [FLINK-17005][docs] Translate the CREATE TABLE ... LIKE syntax documentation to Chinese
flinkbot edited a comment on pull request #12313: URL: https://github.com/apache/flink/pull/12313#issuecomment-633389597 ## CI report: * 07ab8aa279380cc6584c9602a1c344dfb2294074 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2144) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] rkhachatryan commented on pull request #12292: [FLINK-17861][task][checkpointing] Split channel state handles sent to JM
rkhachatryan commented on pull request #12292: URL: https://github.com/apache/flink/pull/12292#issuecomment-633727252 Thanks for the feedback @pnowojski , I've addressed the issues (except [this one](https://github.com/apache/flink/pull/12292#discussion_r429887132)). Answering your question: > Could you elaborate a bit more? What's the alternative? How would it avoid more data duplication? Are we still duplicating data with this PR? Current structure is the following (this PR doesn't change it): ``` Each subtask reports to JM TaskStateSnapshot, each with zero ore more OperatorSubtaskState, each with zero or more InputChannelStateHandle and ResultSubpartitionStateHandle each referencing an underlying StreamStateHandle ``` The underlying `StreamStateHandle` duplicates filename (`ByteStreamStateHandle` has it too at least because of `equals/hashcode` I guess). An alternative would be something like ``` Each subtask reports to JM TaskStateSnapshot, each with zero ore more OperatorSubtaskState, each with zero or one StreamStateHandle for channel state handles each with zero or more InputChannelStateHandle and ResultSubpartitionStateHandle ``` This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] flinkbot edited a comment on pull request #12290: [FLINK-17657][jdbc] jdbc not support read BIGINT UNSIGNED field
flinkbot edited a comment on pull request #12290: URL: https://github.com/apache/flink/pull/12290#issuecomment-632566330 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] EchoLee5 opened a new pull request #12337: [FLINK-17744] StreamContextEnvironment#execute cannot be call JobListener#onJobExecuted
EchoLee5 opened a new pull request #12337: URL: https://github.com/apache/flink/pull/12337 ## What is the purpose of the change If we registered our own joblistener, when job execution ends, onJobExecuted method will be executed ## Brief change log Changes are confined in the StreamContextEnvironment class ## 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, Kubernetes/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) - If yes, how is the feature documented? (JavaDocs) This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] becketqin commented on pull request #12306: [FLINK-17899][runtime] Integrate FLIP-126 Watermarks with FLIP-27 Sources
becketqin commented on pull request #12306: URL: https://github.com/apache/flink/pull/12306#issuecomment-633606359 BTW, I think we do need more tests for the patch. This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] KarmaGYZ commented on pull request #12315: [FLINK-17917][yarn] Ignore the external resource with a value of 0 in…
KarmaGYZ commented on pull request #12315: URL: https://github.com/apache/flink/pull/12315#issuecomment-633776377 Thanks a lot for the review @tillrohrmann . I've updated PR and verified it with Yarn 3.1.0. This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] rmetzger closed pull request #12296: [FLINK-17814][chinese-translation]Translate native kubernetes document to Chinese
rmetzger closed pull request #12296: URL: https://github.com/apache/flink/pull/12296 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] wangyang0918 commented on pull request #12317: [FLINK-17819][yarn] Fix error msg for yarn deployments when hadoop not in classpath
wangyang0918 commented on pull request #12317: URL: https://github.com/apache/flink/pull/12317#issuecomment-634380791 The changes looks really good to me. +1 for merging. This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] wangyang0918 commented on pull request #12296: [FLINK-17814][chinese-translation]Translate native kubernetes document to Chinese
wangyang0918 commented on pull request #12296: URL: https://github.com/apache/flink/pull/12296#issuecomment-633792306 The changes look really good to me now. @klion26 Do you have other concerns? This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink-web] Shadowell opened a new pull request #341: [docs-zh] Translate "stateful-functions" page into Chinese
Shadowell opened a new pull request #341: URL: https://github.com/apache/flink-web/pull/341 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] KarmaGYZ commented on a change in pull request #12315: [FLINK-17917][yarn] Ignore the external resource with a value of 0 in…
KarmaGYZ commented on a change in pull request #12315:
URL: https://github.com/apache/flink/pull/12315#discussion_r430118921
##
File path:
flink-yarn/src/main/java/org/apache/flink/yarn/ResourceInformationReflector.java
##
@@ -151,7 +151,9 @@ void setResourceInformationUnSafe(Object resource, String
resourceName, long amo
for (int i = 2; i < externalResourcesInfo.length; i++) {
final String name = (String)
resourceInformationGetNameMethod.invoke(externalResourcesInfo[i]);
final long value = (long)
resourceInformationGetValueMethod.invoke(externalResourcesInfo[i]);
- externalResources.put(name, value);
+ if (value > 0) {
+ externalResources.put(name, value);
+ }
Review comment:
Yes, you are right.
##
File path:
flink-yarn/src/main/java/org/apache/flink/yarn/ResourceInformationReflector.java
##
@@ -151,7 +151,9 @@ void setResourceInformationUnSafe(Object resource, String
resourceName, long amo
for (int i = 2; i < externalResourcesInfo.length; i++) {
final String name = (String)
resourceInformationGetNameMethod.invoke(externalResourcesInfo[i]);
final long value = (long)
resourceInformationGetValueMethod.invoke(externalResourcesInfo[i]);
- externalResources.put(name, value);
+ if (value > 0) {
+ externalResources.put(name, value);
+ }
Review comment:
Yes, you are right. I think filtering out entries with a value of 0
would be good enough.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to 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
[GitHub] [flink] rmetzger commented on pull request #12251: [FLINK-17812][dist] Bundle reporters in plugins/ directory
rmetzger commented on pull request #12251: URL: https://github.com/apache/flink/pull/12251#issuecomment-634005552 Note, this change is breaking the e2e tests: https://dev.azure.com/georgeryan1322/Flink/_build/results?buildId=326=logs=1f3ed471-1849-5d3c-a34c-19792af4ad16=2f5b54d0-1d28-5b01-d344-aa50ffe0cdf8 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] rmetzger commented on pull request #12340: [FLINK-17844][build] Enforce @PublicEvolving compatibility for minor versions
rmetzger commented on pull request #12340: URL: https://github.com/apache/flink/pull/12340#issuecomment-634201576 I'm not saying you should do it the way I would have done it :) And yeah, it would be a lot of noise in our pom files for such a tiny change. This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] rkhachatryan commented on pull request #12258: [FLINK-17820][task][checkpointing] Don't flush channel state to disk explicitly
rkhachatryan commented on pull request #12258: URL: https://github.com/apache/flink/pull/12258#issuecomment-633844876 Obsoleted by #12332 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] azagrebin commented on a change in pull request #11245: [FLINK-15794][Kubernetes] Generate the Kubernetes default image version
azagrebin commented on a change in pull request #11245:
URL: https://github.com/apache/flink/pull/11245#discussion_r430332353
##
File path: docs/ops/deployment/kubernetes.md
##
@@ -159,6 +159,8 @@ with the `kubectl` command:
## Appendix
### Common cluster resource definitions
+The Deployment definitions use the pre-built image `flink:{% if site.is_stable
%}{{site.version}}-scala{{site.scala_version_suffix}}`{% else %}latest` (which
contains the latest released Flink built against Scala 2.12){% endif %} which
can be found [on Docker Hub](https://hub.docker.com/_/flink/).
+The image is built from this [Github
repository](https://github.com/apache/flink-docker).
Review comment:
We already have a chapter on this page about the image, I would rather
refer to it: `#flink-docker-image`.
This way we would have to change the image description only in one place.
Also, the information about scala 2.12 for latest can become stale, it would
make sense to just refer to the tags page. We have everywhere Scala 2.11 by
default except docker and might want to make it 2.11 for docker as well later.
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/util/EnvironmentInformation.java
##
@@ -103,6 +103,15 @@ public static String getGitCommitTimeString() {
return getVersionsInstance().gitCommitTimeStr;
}
+ /**
+* Returns the exact name of the Dockerimage for Flink that is needed
to run in.
+*
+* @return The "name:tag" of the Flink docker image.
+*/
+ public static String getFlinkDockerImageNameAndTag() {
Review comment:
Why do you think this function belongs to `EnvironmentInformation `?
It looks to me as just a private helper method for `CONTAINER_IMAGE` option
for now.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to 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
[GitHub] [flink] azagrebin closed pull request #12307: [FLINK-15621] Remove deprecated option and method to disable TTL compaction filter
azagrebin closed pull request #12307: URL: https://github.com/apache/flink/pull/12307 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[jira] [Updated] (FLINK-17953) OverWindow doesn't support to order by non-time attribute in batch mode for Table API program
[
https://issues.apache.org/jira/browse/FLINK-17953?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Dian Fu updated FLINK-17953:
Description:
For a simple batch job tested in blink planner:
{code:java}
INSERT INTO results
SELECT id, sum(sales)
OVER (PARTITION BY id ORDER BY ts ROWS BETWEEN 2 PRECEDING AND 0 FOLLOWING)
FROM input
{code}
It could pass if written in SQL. However, if we rewrite it in Table API, it
will throw the following exception:
{code:java}
py4j.protocol.Py4JJavaError: An error occurred while calling
o85.select.py4j.protocol.Py4JJavaError: An error occurred while calling
o85.select.: org.apache.flink.table.api.ValidationException: Ordering must be
defined on a time attribute. at
org.apache.flink.table.planner.expressions.PlannerTypeInferenceUtilImpl.validateArguments(PlannerTypeInferenceUtilImpl.java:112)
at
org.apache.flink.table.planner.expressions.PlannerTypeInferenceUtilImpl.runTypeInference(PlannerTypeInferenceUtilImpl.java:71)
at
org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.runLegacyTypeInference(ResolveCallByArgumentsRule.java:218)
at
org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.lambda$visit$2(ResolveCallByArgumentsRule.java:134)
at java.util.Optional.orElseGet(Optional.java:267) at
org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.visit(ResolveCallByArgumentsRule.java:134)
at
org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.visit(ResolveCallByArgumentsRule.java:89)
at
org.apache.flink.table.expressions.ApiExpressionVisitor.visit(ApiExpressionVisitor.java:39)
at
org.apache.flink.table.expressions.UnresolvedCallExpression.accept(UnresolvedCallExpression.java:132)
at
org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.visit(ResolveCallByArgumentsRule.java:124)
at
org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.visit(ResolveCallByArgumentsRule.java:89)
at
org.apache.flink.table.expressions.ApiExpressionVisitor.visit(ApiExpressionVisitor.java:39)
at
org.apache.flink.table.expressions.UnresolvedCallExpression.accept(UnresolvedCallExpression.java:132)
at
org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule.lambda$apply$0(ResolveCallByArgumentsRule.java:83)
at java.util.stream.ReferencePipeline$7$1.accept(ReferencePipeline.java:267)
at
java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1374)
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481) at
java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471) at
java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708) at
java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) at
java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:499) at
org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule.apply(ResolveCallByArgumentsRule.java:84)
at
org.apache.flink.table.expressions.resolver.ExpressionResolver.lambda$null$1(ExpressionResolver.java:211)
at java.util.function.Function.lambda$andThen$1(Function.java:88) at
org.apache.flink.table.expressions.resolver.ExpressionResolver.resolve(ExpressionResolver.java:178)
at
org.apache.flink.table.operations.utils.OperationTreeBuilder.projectInternal(OperationTreeBuilder.java:191)
at
org.apache.flink.table.operations.utils.OperationTreeBuilder.project(OperationTreeBuilder.java:170)
at
org.apache.flink.table.api.internal.TableImpl$OverWindowedTableImpl.select(TableImpl.java:953)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498) at
org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at
org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282) at
org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at
org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79)
at
org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238)
at java.lang.Thread.run(Thread.java:745)
{code}
was:
For a simple batch job:
{code}
INSERT INTO results
SELECT id, sum(sales)
OVER (PARTITION BY id ORDER BY ts ROWS BETWEEN 2 PRECEDING AND 0 FOLLOWING)
FROM input
{code}
It could pass in blink planner. However, if we rewrite it in Table API, it will
throw the following
[GitHub] [flink] rmetzger commented on pull request #12296: [FLINK-17814][chinese-translation]Translate native kubernetes document to Chinese
rmetzger commented on pull request #12296: URL: https://github.com/apache/flink/pull/12296#issuecomment-633963217 Thanks for the review. Will merge. This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] zentol commented on a change in pull request #12268: [FLINK-17375] Refactor travis_watchdog.sh into separate ci and azure scripts.
zentol commented on a change in pull request #12268:
URL: https://github.com/apache/flink/pull/12268#discussion_r429961569
##
File path: tools/ci/watchdog.sh
##
@@ -0,0 +1,122 @@
+#!/usr/bin/env bash
+
+# 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.
+
+
+#
+# This file contains a watchdog tool to run monitor and potentially kill tasks
+# not producing any output for n seconds.
+#
+
+# Number of seconds w/o output before printing a stack trace and killing the
watched process
+MAX_NO_OUTPUT=${MAX_NO_OUTPUT:-900}
+
+# Number of seconds to sleep before checking the output again
+SLEEP_TIME=${SLEEP_TIME:-20}
+
+CMD_OUT=${CMD_OUT:-"/tmp/watchdog.out"}
+CMD_PID=${CMD_PID:-"/tmp/watchdog.pid"}
+CMD_EXIT=${CMD_EXIT:-"/tmp/watchdog.exit"}
+
+
+# =
+# Utility functions
+# =
+
+mod_time () {
+ echo `stat -c "%Y" $CMD_OUT`
+}
+
+the_time() {
+ echo `date +%s`
+}
+
+# watchdog process
+
+watchdog () {
+ touch $CMD_OUT
+
+ while true; do
+ sleep $SLEEP_TIME
+
+ time_diff=$((`the_time` - `mod_time`))
+
+ if [ $time_diff -ge $MAX_NO_OUTPUT ]; then
+ echo
"=="
+ echo "Process produced no output for ${MAX_NO_OUTPUT}
seconds."
+ echo
"=="
+
+ # run timeout callback
+ $WATCHDOG_CALLBACK_ON_TIMEOUT
+
+ echo "Killing process with pid=$(<$CMD_PID) and all
descendants"
+ pkill -P $(<$CMD_PID) # kill descendants
+ kill $(<$CMD_PID) # kill process itself
+
+ exit 1
+ fi
+ done
+}
+
+assume_available () {
+ VAR=$1
+ if [ -z "$VAR" ] ; then
+ echo "ERROR: Environment variable '$VAR' is not set but
expected by watchdog.sh"
+ exit 1
+ fi
+}
+
+# =
+# main function
+# =
+
+# entrypoint
+function run_with_watchdog() {
+ local cmd="$1"
+
+ # check preconditions
+ assume_available CMD_OUT # used for writing the process output (to
check for activity)
+ assume_available CMD_PID # location of file to write process id to
+ assume_available CMD_EXIT # location of file to writ exit code to
+ assume_available WATCHDOG_CALLBACK_ON_TIMEOUT # bash function to call
on timeout
+
+ watchdog &
Review comment:
It access CMD_PID etc. and the callback.
##
File path: tools/azure-pipelines/jobs-template.yml
##
@@ -121,15 +128,34 @@ jobs:
continueOnError: true # continue the build even if the cache fails.
condition: not(eq('${{parameters.test_pool_definition.name}}', 'Default'))
displayName: Cache Maven local repo
+
- script: |
echo "##vso[task.setvariable variable=JAVA_HOME]$JAVA_HOME_11_X64"
echo "##vso[task.setvariable variable=PATH]$JAVA_HOME_11_X64/bin:$PATH"
displayName: "Set to jdk11"
condition: eq('${{parameters.jdk}}', 'jdk11')
+
- script: sudo sysctl -w kernel.core_pattern=core.%p
displayName: Set coredump pattern
+
# Test
- - script: STAGE=test ${{parameters.environment}}
./tools/azure-pipelines/azure_controller.sh $(module)
+ - script: |
+ ./tools/azure-pipelines/unpack_build_artifact.sh
+ export DEBUG_FILES="$AGENT_TEMPDIRECTORY/debug_files"
Review comment:
I figured as much.
I'd say that if it is not just calling script A,B,C, then add a separate
script for it. What we absolutely want to prevent is the scripts being spread
out over several places.
TBH there's probably even merit in having each step call at most 1 script.
##
File path: tools/azure-pipelines/build-python-wheels.yml
##
@@ -24,7 +24,6
[GitHub] [flink-web] morsapaes commented on a change in pull request #339: [blog] flink on zeppelin
morsapaes commented on a change in pull request #339: URL: https://github.com/apache/flink-web/pull/339#discussion_r429974813 ## File path: _posts/2020-05-25-flink-on-zeppelin.md ## @@ -0,0 +1,83 @@ +--- +layout: post +title: "Flink on Zeppelin Notebooks for Interactive Data Analysis" +date: 2020-05-25T08:00:00.000Z +categories: news +authors: +- zjffdu: + name: "Jeff Zhang" + twitter: "zjffdu" +--- + +The latest release of Apache Zeppelin comes with a redesigned interpreter for Apache Flink (version Flink 1.10+ is only supported moving forward) +that allows developers and data engineers to use Flink directly on Zeppelin notebooks for interactive data analysis. In this post, we explain how the Flink interpreter in Zeppelin works, +and provide a tutorial for running Streaming ETL with Flink on Zeppelin. + +# The Flink Interpreter in Zeppelin 0.9 + +The Flink interpreter can be accessed and configured from Zeppelin’s interpreter settings page. +The interpreter has been refactored so that Flink users can now take advantage of Zeppelin to write Flink applications in three languages, +namely Scala, Python (PyFlink) and SQL (for both batch & streaming executions). +Zeppelin 0.9 now comes with the Flink interpreter group, consisting of the below five interpreters: + +* %flink - Provides a Scala environment +* %flink.pyflink - Provides a python environment +* %flink.ipyflink - Provides an ipython environment +* %flink.bsql - Provides a stream sql environment Review comment: ```suggestion * %flink.ssql - Provides a stream sql environment ``` ## File path: _posts/2020-05-25-flink-on-zeppelin.md ## @@ -0,0 +1,83 @@ +--- +layout: post +title: "Flink on Zeppelin Notebooks for Interactive Data Analysis" +date: 2020-05-25T08:00:00.000Z +categories: news +authors: +- zjffdu: + name: "Jeff Zhang" + twitter: "zjffdu" +--- + +The latest release of Apache Zeppelin comes with a redesigned interpreter for Apache Flink (version Flink 1.10+ is only supported moving forward) +that allows developers and data engineers to use Flink directly on Zeppelin notebooks for interactive data analysis. In this post, we explain how the Flink interpreter in Zeppelin works, +and provide a tutorial for running Streaming ETL with Flink on Zeppelin. + +# The Flink Interpreter in Zeppelin 0.9 + +The Flink interpreter can be accessed and configured from Zeppelin’s interpreter settings page. +The interpreter has been refactored so that Flink users can now take advantage of Zeppelin to write Flink applications in three languages, +namely Scala, Python (PyFlink) and SQL (for both batch & streaming executions). +Zeppelin 0.9 now comes with the Flink interpreter group, consisting of the below five interpreters: + +* %flink - Provides a Scala environment +* %flink.pyflink - Provides a python environment +* %flink.ipyflink - Provides an ipython environment +* %flink.bsql - Provides a stream sql environment +* %flink.ssql - Provides a batch sql environment + +Not only has the interpreter been extended to support writing Flink applications in three languages, but it has also extended the available execution modes for Flink that now include: +* Running Flink in Local Mode Review comment: ```suggestion * Running Flink in Local Mode ``` ## File path: _posts/2020-05-25-flink-on-zeppelin.md ## @@ -0,0 +1,83 @@ +--- +layout: post +title: "Flink on Zeppelin Notebooks for Interactive Data Analysis" +date: 2020-05-25T08:00:00.000Z +categories: news +authors: +- zjffdu: + name: "Jeff Zhang" + twitter: "zjffdu" +--- + +The latest release of Apache Zeppelin comes with a redesigned interpreter for Apache Flink (version Flink 1.10+ is only supported moving forward) +that allows developers and data engineers to use Flink directly on Zeppelin notebooks for interactive data analysis. In this post, we explain how the Flink interpreter in Zeppelin works, +and provide a tutorial for running Streaming ETL with Flink on Zeppelin. + +# The Flink Interpreter in Zeppelin 0.9 + +The Flink interpreter can be accessed and configured from Zeppelin’s interpreter settings page. +The interpreter has been refactored so that Flink users can now take advantage of Zeppelin to write Flink applications in three languages, +namely Scala, Python (PyFlink) and SQL (for both batch & streaming executions). +Zeppelin 0.9 now comes with the Flink interpreter group, consisting of the below five interpreters: + +* %flink - Provides a Scala environment +* %flink.pyflink - Provides a python environment +* %flink.ipyflink - Provides an ipython environment +* %flink.bsql - Provides a stream sql environment +* %flink.ssql - Provides a batch sql environment Review comment: ```suggestion * %flink.bsql - Provides a batch sql environment ``` ## File path: _posts/2020-05-25-flink-on-zeppelin.md
[GitHub] [flink] flinkbot edited a comment on pull request #12264: [FLINK-17558][netty] Release partitions asynchronously
flinkbot edited a comment on pull request #12264: URL: https://github.com/apache/flink/pull/12264#issuecomment-631349883 ## CI report: * 19c5f57b94cc56b70002031618c32d9e6f68effb UNKNOWN * bb313e40f5a72dbf20cd0a8b48267063fd4f00af UNKNOWN * 3510bfd56ae6a431783bbade1881dd967b271457 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=2139) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[jira] [Created] (FLINK-17953) OverWindow doesn't support to order by non-time attribute in batch mode for Table API program
Dian Fu created FLINK-17953:
---
Summary: OverWindow doesn't support to order by non-time attribute
in batch mode for Table API program
Key: FLINK-17953
URL: https://issues.apache.org/jira/browse/FLINK-17953
Project: Flink
Issue Type: Bug
Components: Table SQL / Planner
Affects Versions: 1.10.0, 1.9.0, 1.11.0
Reporter: Dian Fu
For a simple batch job:
{code}
INSERT INTO results
SELECT id, sum(sales)
OVER (PARTITION BY id ORDER BY ts ROWS BETWEEN 2 PRECEDING AND 0 FOLLOWING)
FROM input
{code}
It could pass in blink planner. However, if we rewrite it in Table API, it will
throw the following exception:
{code}
py4j.protocol.Py4JJavaError: An error occurred while calling
o85.select.py4j.protocol.Py4JJavaError: An error occurred while calling
o85.select.: org.apache.flink.table.api.ValidationException: Ordering must be
defined on a time attribute. at
org.apache.flink.table.planner.expressions.PlannerTypeInferenceUtilImpl.validateArguments(PlannerTypeInferenceUtilImpl.java:112)
at
org.apache.flink.table.planner.expressions.PlannerTypeInferenceUtilImpl.runTypeInference(PlannerTypeInferenceUtilImpl.java:71)
at
org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.runLegacyTypeInference(ResolveCallByArgumentsRule.java:218)
at
org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.lambda$visit$2(ResolveCallByArgumentsRule.java:134)
at java.util.Optional.orElseGet(Optional.java:267) at
org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.visit(ResolveCallByArgumentsRule.java:134)
at
org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.visit(ResolveCallByArgumentsRule.java:89)
at
org.apache.flink.table.expressions.ApiExpressionVisitor.visit(ApiExpressionVisitor.java:39)
at
org.apache.flink.table.expressions.UnresolvedCallExpression.accept(UnresolvedCallExpression.java:132)
at
org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.visit(ResolveCallByArgumentsRule.java:124)
at
org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule$ResolvingCallVisitor.visit(ResolveCallByArgumentsRule.java:89)
at
org.apache.flink.table.expressions.ApiExpressionVisitor.visit(ApiExpressionVisitor.java:39)
at
org.apache.flink.table.expressions.UnresolvedCallExpression.accept(UnresolvedCallExpression.java:132)
at
org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule.lambda$apply$0(ResolveCallByArgumentsRule.java:83)
at java.util.stream.ReferencePipeline$7$1.accept(ReferencePipeline.java:267)
at
java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1374)
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481) at
java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471) at
java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708) at
java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) at
java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:499) at
org.apache.flink.table.expressions.resolver.rules.ResolveCallByArgumentsRule.apply(ResolveCallByArgumentsRule.java:84)
at
org.apache.flink.table.expressions.resolver.ExpressionResolver.lambda$null$1(ExpressionResolver.java:211)
at java.util.function.Function.lambda$andThen$1(Function.java:88) at
org.apache.flink.table.expressions.resolver.ExpressionResolver.resolve(ExpressionResolver.java:178)
at
org.apache.flink.table.operations.utils.OperationTreeBuilder.projectInternal(OperationTreeBuilder.java:191)
at
org.apache.flink.table.operations.utils.OperationTreeBuilder.project(OperationTreeBuilder.java:170)
at
org.apache.flink.table.api.internal.TableImpl$OverWindowedTableImpl.select(TableImpl.java:953)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498) at
org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at
org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282) at
org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at
org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79)
at
org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238)
at java.lang.Thread.run(Thread.java:745)
{code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] JingsongLi merged pull request #12314: [FLINK-17756][table-api-java] Drop table/view shouldn't take effect o…
JingsongLi merged pull request #12314: URL: https://github.com/apache/flink/pull/12314 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] zentol commented on pull request #12340: [FLINK-17844][build] Enforce @PublicEvolving compatibility for minor versions
zentol commented on pull request #12340: URL: https://github.com/apache/flink/pull/12340#issuecomment-634127129 A separate profile is imo overkill, when all we really need is to move one entry. This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] klion26 commented on a change in pull request #12296: [FLINK-17814][chinese-translation]Translate native kubernetes document to Chinese
klion26 commented on a change in pull request #12296:
URL: https://github.com/apache/flink/pull/12296#discussion_r430257105
##
File path: docs/ops/deployment/native_kubernetes.zh.md
##
@@ -92,73 +90,73 @@ $ ./bin/kubernetes-session.sh \
-Dkubernetes.container.image=
{% endhighlight %}
-### Submitting jobs to an existing Session
+### 将作业提交到现有 Session
-Use the following command to submit a Flink Job to the Kubernetes cluster.
+使用以下命令将 Flink 作业提交到 Kubernetes 集群。
{% highlight bash %}
$ ./bin/flink run -d -e kubernetes-session -Dkubernetes.cluster-id=
examples/streaming/WindowJoin.jar
{% endhighlight %}
-### Accessing Job Manager UI
+### 访问 Job Manager UI
-There are several ways to expose a Service onto an external (outside of your
cluster) IP address.
-This can be configured using `kubernetes.service.exposed.type`.
+有几种方法可以将服务暴露到外部（集群外部） IP 地址。
+可以使用 `kubernetes.service.exposed.type` 进行配置。
-- `ClusterIP`: Exposes the service on a cluster-internal IP.
-The Service is only reachable within the cluster. If you want to access the
Job Manager ui or submit job to the existing session, you need to start a local
proxy.
-You can then use `localhost:8081` to submit a Flink job to the session or view
the dashboard.
+- `ClusterIP`：通过集群内部 IP 暴露服务。
+该服务只能在集群中访问。如果想访问 JobManager ui 或将作业提交到现有 session，则需要启动一个本地代理。
+然后你可以使用 `localhost:8081` 将 Flink 作业提交到 session 或查看仪表盘。
{% highlight bash %}
$ kubectl port-forward service/ 8081
{% endhighlight %}
-- `NodePort`: Exposes the service on each Node’s IP at a static port (the
`NodePort`). `:` could be used to contact the Job Manager
Service. `NodeIP` could be easily replaced with Kubernetes ApiServer address.
-You could find it in your kube config file.
+- `NodePort`：通过每个 Node 上的 IP 和静态端口（`NodePort`）暴露服务。`:`
可以用来连接 JobManager 服务。`NodeIP` 可以很容易地用 Kubernetes ApiServer 地址替换。
+你可以在 kube 配置文件找到它。
-- `LoadBalancer`: Default value, exposes the service externally using a cloud
provider’s load balancer.
-Since the cloud provider and Kubernetes needs some time to prepare the load
balancer, you may get a `NodePort` JobManager Web Interface in the client log.
-You can use `kubectl get services/` to get EXTERNAL-IP and then
construct the load balancer JobManager Web Interface manually
`http://:8081`.
+- `LoadBalancer`：默认值，使用云提供商的负载均衡器在外部暴露服务。
+由于云提供商和 Kubernetes 需要一些时间来准备负载均衡器，因此你可以在客户端日志中获得一个 `NodePort` 的 JobManager Web
界面。
+你可以使用 `kubectl get services/`获取 EXTERNAL-IP 然后手动构建负载均衡器 JobManager
Web 界面 `http://:8081`。
Review comment:
这个还是建议加上空格，这个 `kubectl get services/`
是一个整体，和后面的中文隔离开，排版之后会更好一些
##
File path: docs/ops/deployment/native_kubernetes.zh.md
##
@@ -24,43 +24,41 @@ specific language governing permissions and limitations
under the License.
-->
-This page describes how to deploy a Flink session cluster natively on
[Kubernetes](https://kubernetes.io).
+本页面描述了如何在 [Kubernetes](https://kubernetes.io) 原生地部署 Flink session 集群。
* This will be replaced by the TOC
{:toc}
-Flink's native Kubernetes integration is still experimental. There may be
changes in the configuration and CLI flags in latter versions.
+Flink 的原生 Kubernetes 集成仍处于试验阶段。在以后的版本中，配置和 CLI flags 可能会发生变化。
-## Requirements
+## 要求
-- Kubernetes 1.9 or above.
-- KubeConfig, which has access to list, create, delete pods and services,
configurable via `~/.kube/config`. You can verify permissions by running
`kubectl auth can-i pods`.
-- Kubernetes DNS enabled.
-- A service Account with [RBAC](#rbac) permissions to create, delete pods.
+- Kubernetes 版本 1.9 或以上。
+- KubeConfig 可以查看、创建、删除 pods 和 services，可以通过`~/.kube/config` 配置。你可以通过运行
`kubectl auth can-i pods` 来验证权限。
+- 启用 Kubernetes DNS。
+- 具有 [RBAC](#rbac) 权限的 Service Account 可以创建、删除 pods。
## Flink Kubernetes Session
-### Start Flink Session
+### 启动 Flink Session
-Follow these instructions to start a Flink Session within your Kubernetes
cluster.
+按照以下说明在 Kubernetes 集群中启动 Flink Session。
-A session will start all required Flink services (JobManager and TaskManagers)
so that you can submit programs to the cluster.
-Note that you can run multiple programs per session.
+Session 集群将启动所有必需的 Flink 服务（JobManager 和 TaskManagers），以便你可以将程序提交到集群。
+注意你可以在每个 session 上运行多个程序。
{% highlight bash %}
$ ./bin/kubernetes-session.sh
{% endhighlight %}
-All the Kubernetes configuration options can be found in our [configuration
guide]({{ site.baseurl }}/zh/ops/config.html#kubernetes).
+所有 Kubernetes 配置项都可以在我们的[配置指南]({{ site.baseurl
}}/zh/ops/config.html#kubernetes)中找到。
-**Example**: Issue the following command to start a session cluster with 4 GB
of memory and 2 CPUs with 4 slots per TaskManager:
+**示例**: 执行以下命令启动 session 集群，每个 TaskManager 分配 4 GB 内存、2 CPUs、4 slots：
-In this example we override the `resourcemanager.taskmanager-timeout` setting
to make
-the pods with task managers remain for a longer period than the default of 30
seconds.
-Although this setting may cause more cloud
[GitHub] [flink] flinkbot commented on pull request #12323: [FLINK-17925][fs-connector] Fix Filesystem options to default values and types
flinkbot commented on pull request #12323: URL: https://github.com/apache/flink/pull/12323#issuecomment-633592742 ## CI report: * c22d8ec0765bdfe50423a5a80d6641129f94e544 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
[GitHub] [flink] flinkbot edited a comment on pull request #12078: [FLINK-17610][state] Align the behavior of result of internal map state to return empty iterator
flinkbot edited a comment on pull request #12078: URL: https://github.com/apache/flink/pull/12078#issuecomment-626611802 This is an automated message from the Apache Git Service. To respond to the message, please log on to 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
