[jira] [Created] (FLINK-7011) Instable Kafka testStartFromKafkaCommitOffsets failures on Travis
Tzu-Li (Gordon) Tai created FLINK-7011:
--
Summary: Instable Kafka testStartFromKafkaCommitOffsets failures
on Travis
Key: FLINK-7011
URL: https://issues.apache.org/jira/browse/FLINK-7011
Project: Flink
Issue Type: Bug
Components: Kafka Connector, Tests
Affects Versions: 1.3.1, 1.4.0
Reporter: Tzu-Li (Gordon) Tai
Assignee: Tzu-Li (Gordon) Tai
Example:
https://s3.amazonaws.com/archive.travis-ci.org/jobs/246703474/log.txt?X-Amz-Expires=30&X-Amz-Date=20170627T065647Z&X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=AKIAJRYRXRSVGNKPKO5A/20170627/us-east-1/s3/aws4_request&X-Amz-SignedHeaders=host&X-Amz-Signature=dbfc90cfc386fef0990325b54ff74ee4d441944687e7fdaa73ce7b0c2b2ec0ea
In general, the test {{testStartFromKafkaCommitOffsets}} implementation is a
bit of an overkill. Before continuing with the test, it writes some records
just for the sake of committing offsets to Kafka and waits for some offsets to
be committed (which leads to the instability), whereas we can do that simply
using the test base's {{OffsetHandler}}.
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
Re: [DISCUSS] Backwards compatibility policy.
For many parts of the code, I would agree with Aljoscha. However, I can also see notable exceptions, such as maintaining support for the legacy state from Flink <=1.1. For example, I think dropping support for this can simplify new developments such as fast local recovery or state replication quiet a bit because this is a special case that runs through a lot of code from backend to JM. So besides this general discussion about a backwards compatible policy, do you think it could make sense to start another concrete discussion about if we still must or want backwards compatibility to Flink 1.1 in Flink 1.4? > Am 29.05.2017 um 12:08 schrieb Aljoscha Krettek : > > Normally, I’m the first one to suggest removing everything that is not > absolutely necessary in order to have a clean code base. On this issue, > though, I think we should support restoring from old Savepoints as far back > as possible if it does not make the code completely unmaintainable. Some > users might jump versions and always forcing them to go though every version > from their old version to the current version doesn’t seem feasible and might > put off some users. > > So far, I think the burden of supporting restore from 1.1 is still small > enough and with each new version the changes between versions become less and > less. The changes from 1.2 to the upcoming 1.3 are quite minimal, I think. > > Best, > Aljoscha >> On 24. May 2017, at 17:58, Ted Yu wrote: >> >> bq. about having LTS versions once a year >> >> +1 to the above. >> >> There may be various reasons users don't want to upgrade (after new >> releases come out). We should give such users enough flexibility on the >> upgrade path. >> >> Cheers >> >> On Wed, May 24, 2017 at 8:39 AM, Kostas Kloudas >> wrote: >> >>> Hi all, >>> >>> For the proposal of having a third party tool, I agree with Ted. >>> Maintaining >>> it is a big and far from trivial effort. >>> >>> Now for the window of backwards compatibility, I would argue that even if >>> for some users 4 months (1 release) is not enough to bump their Flink >>> version, >>> the proposed policy guarantees that there will always be a path from any >>> old >>> version to any subsequent one. >>> >>> Finally, for the proposal about having LTS versions once a year, I am not >>> sure if this will reduce or create more overhead. If I understand the plan >>> correctly, this would mean that the community will have to maintain >>> 2 or 3 LTS versions and the last two major ones, right? >>> On May 22, 2017, at 7:31 PM, Ted Yu wrote: For #2, it is difficult to achieve: a. maintaining savepoint migration is non-trivial and should be reviewed >>> by domain experts b. how to certify such third-party tool Cheers On Mon, May 22, 2017 at 3:04 AM, 施晓罡 wrote: > Hi all, > > Currently, we work a lot in the maintenance of compatibility. > There exist much code in runtime to support the migration of savepoints > (most of which are deprecated), making it hard to focus on the current > implementation. > When more versions are released, much more efforts will be needed if we > try to make these released versions compatible. > > I agree with Tzu-Li that we should provide a method to let users upgrade > Flink in a reasonable pace. > But i am against the proposal that we only offer backwards compatibility > for one previous version. > According our time-based release model, a major version is released >>> every > four month. > That means, users have to upgrade their versions every 8 months. >>> Otherwise > they will have difficulties in the migration of existing savepoints. > > My suggestions include > > (1) We can release Long-Term Support (LTS) versions which are widely > adopted in other open-source projects. > LTS versions should be stable and are free of found bugs. Savepoints in > LTS versions are guaranteed to be back-compatible so that users can >>> easily > upgrade to newer LTS versions. > > The releasing of LTS versions is slower than that of major versions >>> (maybe > once a year, determined by users’ upgrade frequency). > Each LTS version will be supported a period of time and typically there > are no more than three active LTS versions. > By encouraging users to use LTS versions, we can ease the maintenance of > released versions (bug fixes, back compatibility, and critical >>> performance > improvement). > > (2) We can provide a third-party tool to do the migration of >>> old-versioned > savepoints. > When users upgrade their versions, they can use the provided tool to > migrate existing savepoints. > This can help move the code for savepoint migration out of the actual > codebase, making code focuses on current implementation. > > What do you think? > > Regards, > Xi
[jira] [Created] (FLINK-7012) remove user-JAR upload when disposing a savepoint the old way
Nico Kruber created FLINK-7012:
--
Summary: remove user-JAR upload when disposing a savepoint the old
way
Key: FLINK-7012
URL: https://issues.apache.org/jira/browse/FLINK-7012
Project: Flink
Issue Type: Bug
Components: Client, State Backends, Checkpointing
Affects Versions: 1.4.0
Reporter: Nico Kruber
Assignee: Nico Kruber
Priority: Minor
Inside {{CliFrontend#disposeSavepoint()}}, user JAR files are being uploaded to
the {{BlobServer}} but they are actually not used (also also not cleaned up) in
the job manager's handling of the {{DisposeSavepoint}} message.
Since removing new savepoints is as simple as deleting files and old savepoints
have always worked without these user JARs, we should remove the upload to be
able to make the JAR file upload jobId-dependent for FLINK-6916.
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
[GitHub] flink-shaded pull request #3: [FLINK-7007] Add README.md
GitHub user zentol opened a pull request: https://github.com/apache/flink-shaded/pull/3 [FLINK-7007] Add README.md You can merge this pull request into a Git repository by running: $ git pull https://github.com/zentol/flink-shaded 7007 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink-shaded/pull/3.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #3 commit e3e2588cd3f714dba16cfe328eb22f018bf1e1b4 Author: zentol Date: 2017-06-27T10:44:23Z [FLINK-7007] Add README.md --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[jira] [Created] (FLINK-7013) Add shaded netty dependency
Chesnay Schepler created FLINK-7013: --- Summary: Add shaded netty dependency Key: FLINK-7013 URL: https://issues.apache.org/jira/browse/FLINK-7013 Project: Flink Issue Type: Sub-task Components: Network Reporter: Chesnay Schepler Assignee: Chesnay Schepler -- This message was sent by Atlassian JIRA (v6.4.14#64029)
[jira] [Created] (FLINK-7014) Expose isDeterministic interface to ScalarFunction and TableFunction
Ruidong Li created FLINK-7014: - Summary: Expose isDeterministic interface to ScalarFunction and TableFunction Key: FLINK-7014 URL: https://issues.apache.org/jira/browse/FLINK-7014 Project: Flink Issue Type: Improvement Components: Table API & SQL Reporter: Ruidong Li Currently, the `isDeterministic` method of implementations of `SqlFuntion` are always returning true, which cause inappropriate optimization in Calcite, such as taking user's stateful UDF as a pure functional procedure. -- This message was sent by Atlassian JIRA (v6.4.14#64029)
[jira] [Created] (FLINK-7015) Separate OperatorConfig from StreamConfig
Xu Pingyong created FLINK-7015: -- Summary: Separate OperatorConfig from StreamConfig Key: FLINK-7015 URL: https://issues.apache.org/jira/browse/FLINK-7015 Project: Flink Issue Type: Improvement Reporter: Xu Pingyong Assignee: Xu Pingyong Now stream config contains configs not only the batch task needs, but also the operator needs, so stream config can see configs of the operator, and operator can see configs of the batch task. We need to separate operator config from stream config, and they can only see configs of themselves. -- This message was sent by Atlassian JIRA (v6.4.14#64029)
[jira] [Created] (FLINK-7016) Move inputFormat to InputFormatVertex from TaskConfig
Xu Pingyong created FLINK-7016: -- Summary: Move inputFormat to InputFormatVertex from TaskConfig Key: FLINK-7016 URL: https://issues.apache.org/jira/browse/FLINK-7016 Project: Flink Issue Type: Improvement Reporter: Xu Pingyong Assignee: Xu Pingyong On batch case, InputFormat is put into TaskConfig, batch task gets it to read data and job manager uses it to split splits from TaskConfig. On streaming case, all configs are put into StreamConfig, but this inputFormat is put into TaskConfig. We can put InputFormat into InputFormatVertex, and batch task still gets InputFormat from TaskConfig. It will be clear. -- This message was sent by Atlassian JIRA (v6.4.14#64029)
[jira] [Created] (FLINK-7017) Remove netty usages in flink-tests
Chesnay Schepler created FLINK-7017: --- Summary: Remove netty usages in flink-tests Key: FLINK-7017 URL: https://issues.apache.org/jira/browse/FLINK-7017 Project: Flink Issue Type: Improvement Components: Tests Affects Versions: 1.4.0 Reporter: Chesnay Schepler Assignee: Chesnay Schepler Fix For: 1.4.0 -- This message was sent by Atlassian JIRA (v6.4.14#64029)
[jira] [Created] (FLINK-7018) Reconstruct streamgraph to clear interface
Xu Pingyong created FLINK-7018: -- Summary: Reconstruct streamgraph to clear interface Key: FLINK-7018 URL: https://issues.apache.org/jira/browse/FLINK-7018 Project: Flink Issue Type: Improvement Reporter: Xu Pingyong Assignee: Xu Pingyong StreamGraph not only contains streamNodes and streamEdges, but also contains virtual nodes who has nothing to do with the streamGraph. Virtual nodes should be converted to streamNodes in StreamGraphGenerator.class -- This message was sent by Atlassian JIRA (v6.4.14#64029)
[GitHub] flink-shaded pull request #4: Add flink-shaded-netty module
GitHub user zentol opened a pull request: https://github.com/apache/flink-shaded/pull/4 Add flink-shaded-netty module You can merge this pull request into a Git repository by running: $ git pull https://github.com/zentol/flink-shaded 7013 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink-shaded/pull/4.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #4 commit 85465f2909e40952432ac853e24ba73fa90ff73c Author: zentol Date: 2017-06-27T12:35:52Z Add flink-shaded-netty module --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[jira] [Created] (FLINK-7019) Rework parallelism in Gelly algorithms and examples
Greg Hogan created FLINK-7019:
-
Summary: Rework parallelism in Gelly algorithms and examples
Key: FLINK-7019
URL: https://issues.apache.org/jira/browse/FLINK-7019
Project: Flink
Issue Type: Sub-task
Components: Gelly
Affects Versions: 1.4.0
Reporter: Greg Hogan
Assignee: Greg Hogan
Priority: Minor
Fix For: 1.4.0
Flink job parallelism is set with {{ExecutionConfig#setParallelism}} or when
{{-p}} on the command-line. The Gelly algorithms {{JaccardIndex}},
{{AdamicAdar}}, {{TriangleListing}}, and {{ClusteringCoefficient}} have
intermediate operators which generated output quadratic in the size of input.
These algorithms may need to be run with a high parallelism but doing so for
all operations is wasteful. Thus was introduced "little parallelism".
This can be simplified by moving the parallelism parameter to the new common
base class and with the rule-of-thumb to use the algorithm parallelism for all
normal (small output) operators. The asymptotically large operators will
default to the job parallelism, as will the default algorithm parallelism.
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
Re: [DISCUSS] Backwards compatibility policy.
I think that this discussion is probably motivated especially by the "legacy state" handling of Flink 1.1. The biggest gain in codebase and productivity would be won only by dropping 1.1 compatibility in Flink 1.4. My gut feeling is that this is reasonable. We support two versions back, which means that users can skip one upgrade, but not two. >From what I can tell, users are usually eager to upgrade. They don't do it immediately, but as soon as the new release is a bit battle tested. I would expect skipping two entire versions to be rare enough to be okay with a solution which is a bit more effort for the user: You can upgrade from Flink 1.1. to 1.4 by loading the 1.1 savepoint into Flink 1.2, take a savepoint (1.2 format), and resume that in Flink 1.4. Greetings, Stephan On Tue, Jun 27, 2017 at 12:01 PM, Stefan Richter < s.rich...@data-artisans.com> wrote: > For many parts of the code, I would agree with Aljoscha. However, I can > also see notable exceptions, such as maintaining support for the legacy > state from Flink <=1.1. For example, I think dropping support for this can > simplify new developments such as fast local recovery or state replication > quiet a bit because this is a special case that runs through a lot of code > from backend to JM. So besides this general discussion about a backwards > compatible policy, do you think it could make sense to start another > concrete discussion about if we still must or want backwards compatibility > to Flink 1.1 in Flink 1.4? > > > Am 29.05.2017 um 12:08 schrieb Aljoscha Krettek : > > > > Normally, I’m the first one to suggest removing everything that is not > absolutely necessary in order to have a clean code base. On this issue, > though, I think we should support restoring from old Savepoints as far back > as possible if it does not make the code completely unmaintainable. Some > users might jump versions and always forcing them to go though every > version from their old version to the current version doesn’t seem feasible > and might put off some users. > > > > So far, I think the burden of supporting restore from 1.1 is still small > enough and with each new version the changes between versions become less > and less. The changes from 1.2 to the upcoming 1.3 are quite minimal, I > think. > > > > Best, > > Aljoscha > >> On 24. May 2017, at 17:58, Ted Yu wrote: > >> > >> bq. about having LTS versions once a year > >> > >> +1 to the above. > >> > >> There may be various reasons users don't want to upgrade (after new > >> releases come out). We should give such users enough flexibility on the > >> upgrade path. > >> > >> Cheers > >> > >> On Wed, May 24, 2017 at 8:39 AM, Kostas Kloudas < > k.klou...@data-artisans.com > >>> wrote: > >> > >>> Hi all, > >>> > >>> For the proposal of having a third party tool, I agree with Ted. > >>> Maintaining > >>> it is a big and far from trivial effort. > >>> > >>> Now for the window of backwards compatibility, I would argue that even > if > >>> for some users 4 months (1 release) is not enough to bump their Flink > >>> version, > >>> the proposed policy guarantees that there will always be a path from > any > >>> old > >>> version to any subsequent one. > >>> > >>> Finally, for the proposal about having LTS versions once a year, I am > not > >>> sure if this will reduce or create more overhead. If I understand the > plan > >>> correctly, this would mean that the community will have to maintain > >>> 2 or 3 LTS versions and the last two major ones, right? > >>> > On May 22, 2017, at 7:31 PM, Ted Yu wrote: > > For #2, it is difficult to achieve: > > a. maintaining savepoint migration is non-trivial and should be > reviewed > >>> by > domain experts > b. how to certify such third-party tool > > Cheers > > On Mon, May 22, 2017 at 3:04 AM, 施晓罡 wrote: > > > Hi all, > > > > Currently, we work a lot in the maintenance of compatibility. > > There exist much code in runtime to support the migration of > savepoints > > (most of which are deprecated), making it hard to focus on the > current > > implementation. > > When more versions are released, much more efforts will be needed if > we > > try to make these released versions compatible. > > > > I agree with Tzu-Li that we should provide a method to let users > upgrade > > Flink in a reasonable pace. > > But i am against the proposal that we only offer backwards > compatibility > > for one previous version. > > According our time-based release model, a major version is released > >>> every > > four month. > > That means, users have to upgrade their versions every 8 months. > >>> Otherwise > > they will have difficulties in the migration of existing savepoints. > > > > My suggestions include > > > > (1) We can release Long-Term Support (LTS) versions which are widely > > adopted in other open-source projects. > > LTS vers
[jira] [Created] (FLINK-7020) Upgrade calcite to calcite 1.13
sunjincheng created FLINK-7020: -- Summary: Upgrade calcite to calcite 1.13 Key: FLINK-7020 URL: https://issues.apache.org/jira/browse/FLINK-7020 Project: Flink Issue Type: Improvement Components: Table API & SQL Affects Versions: 1.4.0 Reporter: sunjincheng Assignee: sunjincheng Calcite 1.13 has released, So I'll try to upgrade the dependence. -- This message was sent by Atlassian JIRA (v6.4.14#64029)
[GitHub] flink-shaded issue #4: Add flink-shaded-netty module
Github user rmetzger commented on the issue: https://github.com/apache/flink-shaded/pull/4 +1 to merge --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink-shaded pull request #3: [FLINK-7007] Add README.md
Github user rmetzger commented on a diff in the pull request: https://github.com/apache/flink-shaded/pull/3#discussion_r124335176 --- Diff: README.md --- @@ -0,0 +1,28 @@ + + +# Apache Flink Shaded Dependencies + +This repository contains a number of shaded dependencies for the [Apache Flink](https://flink.apache.org/) project. + +The purpose of these dependencies is to provide a single instance of a shaded dependency in the Flink distribution, instead of each individual module shading the dependency. + +The shaded dependencies contained here do not expose any transitive dependencies. They may or may not be self-contained. + +When using these dependencies it is recommended to work directly against the shaded namespaces. --- End diff -- Could you add the following ``` About Apache Flink is an open source project of The Apache Software Foundation (ASF). ``` and link to the ASF? Other than that +1 to merge --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[DISCUSS] FLIP-21 - Improve object Copying/Reuse Mode for Streaming Runtime
Hi all!
I would like to propose the following FLIP:
FLIP-21 - Improve object Copying/Reuse Mode for Streaming Runtime:
https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=71012982
The FLIP is motivated by the fact that many users run into an unnecessary
kind of performance problem caused by an old design artifact.
The required change should be reasonably small, and would help many users
and Flink's general standing.
Happy to hear thoughts!
Stephan
==
FLIP text is below. Pictures with illustrations are only in the Wiki, not
supported on the mailing list.
-
Motivation
The default behavior of the streaming runtime is to copy every element
between chained operators.
That operation was introduced for “safety” reasons, to avoid the number of
cases where users can create incorrect programs by reusing mutable objects
(a discouraged pattern, but possible). For example when using state
backends that keep the state as objects on heap, reusing mutable objects
can theoretically create cases where the same object is used in multiple
state mappings.
The effect is that many people that try Flink get much lower performance
than they could possibly get. From empirical evidence, almost all users
that I (Stephan) have been in touch with eventually run into this issue
eventually.
There are multiple observations about that design:
-
Object copies are extremely costly. While some simple copy virtually for
free (types reliably detected as immutable are not copied at all), many
real pipelines use types like Avro, Thrift, JSON, etc, which are very
expensive to copy.
-
Keyed operations currently only occur after shuffles. The operations are
hence the first in a pipeline and will never have a reused object anyways.
That means for the most critical operation, this pre-caution is unnecessary.
-
The mode is inconsistent with the contract of the DataSet API, which
does not copy at each step
-
To prevent these copies, users can select {{enableObjectReuse()}}, which
is misleading, since it does not really reuse mutable objects, but avoids
additional copies.
Proposal
Summary
I propose to change the default behavior of the DataStream runtime to be
the same as the DataSet runtime. That means that new objects are chosen on
every deserialization, and no copies are made as the objects are passed on
along the pipelines.
Details
I propose to drop the execution config flag {{objectReuse}} and instead
introduce an {{ObjectReuseMode}} enumeration with better control of what
should happen. There will be three different types:
-
DEFAULT
-
This is the default in the DataSet API
-
This will become the default in the DataStream API
-
This happens in the DataStream API when {{enableObjectReuse()}} is
activated.
-
COPY_PER_OPERATOR
-
The current default in the DataStream API
-
FULL_REUSE
-
This happens in the DataSet API when {{enableObjectReuse()}} is
chosen.
An illustration of the modes is as follows:
DEFAULT
See here:
https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=71012982&preview=/https%3A%2F%2Flh5.googleusercontent.com%2F1UOpVB2wSMhx8067IE9t2_mJG549IoOkDiAfIN_uXQZVUvAXCp-hQLY-mgoSWunwF-xciZuJ4pZpj1FX0ZPQrd-Fm1jWzgX3Hv7-SELUdPUvEN6XUPbLrwfA9YRl605bFKMYlf1r
COPY_PER_OPERATOR
See here:
https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=71012982&preview=/https%3A%2F%2Flh3.googleusercontent.com%2Fs5sBOktzaKrRw3v1-IQMgImYZfchQMVz2HiG3i050xCWNTKuQV6mmlv3QtR0TZ0SGPRSCyjI-sUAqfbJw4fGOxKqBuRX2f-iZGh0e7hBke7DzuApUNy1vaF2SgtQVH3XEXkRx8Ks
FULL_REUSE
See here:
https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=71012982&preview=/https%3A%2F%2Flh5.googleusercontent.com%2FFdOzuuaioooEIOh7bo0gZ5dHZrlgEKiwtNjGE9DjR-fT20B0q7FGDAvAk5oh1h58WtNQktuFGinrV1q1Yq8H8ayCyyqFUq-gmAYYW91x4XZQNrjLc6eJ0cptzvN_r8cU_GVV7LNE
New or Changed Public Interfaces
Interfaces changed
The interface of the {{ExecutionConfig}} add the method
{{setObjectReuseMode(ObjectReuseMode)}}, and deprecates the methods
{{enableObjectReuse()}} and {{disableObjectReuse()}}.
Behavior changed
The default object passing behavior changes, meaning that it can affect the
correctness of prior DataStream programs that assume the original
“COPY_PER_OPERATOR” behavior.
Migration Plan and Compatibility
Interfaces
No interface migration path is needed, because the interfaces are not
broken, merely some methods get deprecated.
Behavior change
Variant 1:
-
Change the behavior, make it explicit on the release notes that we did
that and what cases are affected.
-
This may actually be feasible, because the cases that are affected are
quite pathological corner cases that only very bad implementations should
encounter (se
Re: [DISCUSS] FLIP-21 - Improve object Copying/Reuse Mode for Streaming Runtime
Hi Stephan,
Would this be an appropriate time to discuss allowing reuse to be a
per-operator configuration? Object reuse for chained operators has lead to
considerable surprise for some users of the DataSet API. This came up during
the rework of the object reuse documentation for the DataSet API. With
annotations a Function could mark whether input/iterator or output/collected
objects should be copied or reused.
My distant observation is that is is safer to locally assert reuse at the
operator level than to assume or guarantee the safety of object reuse across an
entire program. It could also be handy to mix operators receiving copyable
objects with operators not requiring copyable objects.
Greg
> On Jun 27, 2017, at 1:21 PM, Stephan Ewen wrote:
>
> Hi all!
>
> I would like to propose the following FLIP:
>
> FLIP-21 - Improve object Copying/Reuse Mode for Streaming Runtime:
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=71012982
>
> The FLIP is motivated by the fact that many users run into an unnecessary
> kind of performance problem caused by an old design artifact.
>
> The required change should be reasonably small, and would help many users
> and Flink's general standing.
>
> Happy to hear thoughts!
>
> Stephan
>
> ==
>
> FLIP text is below. Pictures with illustrations are only in the Wiki, not
> supported on the mailing list.
> -
>
> Motivation
>
> The default behavior of the streaming runtime is to copy every element
> between chained operators.
>
> That operation was introduced for “safety” reasons, to avoid the number of
> cases where users can create incorrect programs by reusing mutable objects
> (a discouraged pattern, but possible). For example when using state
> backends that keep the state as objects on heap, reusing mutable objects
> can theoretically create cases where the same object is used in multiple
> state mappings.
>
> The effect is that many people that try Flink get much lower performance
> than they could possibly get. From empirical evidence, almost all users
> that I (Stephan) have been in touch with eventually run into this issue
> eventually.
>
> There are multiple observations about that design:
>
>
> -
>
> Object copies are extremely costly. While some simple copy virtually for
> free (types reliably detected as immutable are not copied at all), many
> real pipelines use types like Avro, Thrift, JSON, etc, which are very
> expensive to copy.
>
>
>
> -
>
> Keyed operations currently only occur after shuffles. The operations are
> hence the first in a pipeline and will never have a reused object anyways.
> That means for the most critical operation, this pre-caution is unnecessary.
>
>
>
> -
>
> The mode is inconsistent with the contract of the DataSet API, which
> does not copy at each step
>
>
>
> -
>
> To prevent these copies, users can select {{enableObjectReuse()}}, which
> is misleading, since it does not really reuse mutable objects, but avoids
> additional copies.
>
>
> Proposal
>
> Summary
>
> I propose to change the default behavior of the DataStream runtime to be
> the same as the DataSet runtime. That means that new objects are chosen on
> every deserialization, and no copies are made as the objects are passed on
> along the pipelines.
>
> Details
>
> I propose to drop the execution config flag {{objectReuse}} and instead
> introduce an {{ObjectReuseMode}} enumeration with better control of what
> should happen. There will be three different types:
>
>
> -
>
> DEFAULT
> -
>
> This is the default in the DataSet API
> -
>
> This will become the default in the DataStream API
> -
>
> This happens in the DataStream API when {{enableObjectReuse()}} is
> activated.
>
>
>
> -
>
> COPY_PER_OPERATOR
> -
>
> The current default in the DataStream API
>
>
>
> -
>
> FULL_REUSE
> -
>
> This happens in the DataSet API when {{enableObjectReuse()}} is
> chosen.
>
>
> An illustration of the modes is as follows:
>
> DEFAULT
>
>
> See here:
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=71012982&preview=/https%3A%2F%2Flh5.googleusercontent.com%2F1UOpVB2wSMhx8067IE9t2_mJG549IoOkDiAfIN_uXQZVUvAXCp-hQLY-mgoSWunwF-xciZuJ4pZpj1FX0ZPQrd-Fm1jWzgX3Hv7-SELUdPUvEN6XUPbLrwfA9YRl605bFKMYlf1r
>
> COPY_PER_OPERATOR
>
>
> See here:
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=71012982&preview=/https%3A%2F%2Flh3.googleusercontent.com%2Fs5sBOktzaKrRw3v1-IQMgImYZfchQMVz2HiG3i050xCWNTKuQV6mmlv3QtR0TZ0SGPRSCyjI-sUAqfbJw4fGOxKqBuRX2f-iZGh0e7hBke7DzuApUNy1vaF2SgtQVH3XEXkRx8Ks
>
>
> FULL_REUSE
>
>
> See here:
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=71012982&preview=/https%3A%2F%2Flh5.googleusercontent.com%2FFdOzuuaioooEIOh
Re: [DISCUSS] FLIP-21 - Improve object Copying/Reuse Mode for Streaming Runtime
Stephan,
Fully supporting this FLIP. We originally encountered pretty big surprises
observing the object copy behavior causing significant performance degradation
for our massively parallel use case.
In our case, (I think most appropriately SHOULD be the assumptions for all
streaming use case), is to assume object immutability for all the records
throughout processing pipeline, as a result, I don't see a need to distinguish
different object reuse behaviors for different (chained) operators (or to the
very extreme even the need to support COPY_PER_OPERATOR other than we probably
have to support for backward compatibility). I am also a fan of failing fast if
user asserts incorrect assumptions.
One feedback on the FLIP-21 itself, I am not very clear on the difference
between DEFAULT and FULL_REUSE enumeration, aren't them exactly the same thing
in new proposal? However, the model figures seem to indicate they are slightly
different? Can you elaborate a bit more?
Z.
On 2017-06-27 11:14 (-0700), Greg Hogan wrote:
> Hi Stephan,
>
> Would this be an appropriate time to discuss allowing reuse to be a
> per-operator configuration? Object reuse for chained operators has lead to
> considerable surprise for some users of the DataSet API. This came up during
> the rework of the object reuse documentation for the DataSet API. With
> annotations a Function could mark whether input/iterator or output/collected
> objects should be copied or reused.
>
> My distant observation is that is is safer to locally assert reuse at the
> operator level than to assume or guarantee the safety of object reuse across
> an entire program. It could also be handy to mix operators receiving copyable
> objects with operators not requiring copyable objects.
>
> Greg
>
>
> > On Jun 27, 2017, at 1:21 PM, Stephan Ewen wrote:
> >
> > Hi all!
> >
> > I would like to propose the following FLIP:
> >
> > FLIP-21 - Improve object Copying/Reuse Mode for Streaming Runtime:
> > https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=71012982
> >
> > The FLIP is motivated by the fact that many users run into an unnecessary
> > kind of performance problem caused by an old design artifact.
> >
> > The required change should be reasonably small, and would help many users
> > and Flink's general standing.
> >
> > Happy to hear thoughts!
> >
> > Stephan
> >
> > ==
> >
> > FLIP text is below. Pictures with illustrations are only in the Wiki, not
> > supported on the mailing list.
> > -
> >
> > Motivation
> >
> > The default behavior of the streaming runtime is to copy every element
> > between chained operators.
> >
> > That operation was introduced for âsafetyâ reasons, to avoid the number
> > of
> > cases where users can create incorrect programs by reusing mutable objects
> > (a discouraged pattern, but possible). For example when using state
> > backends that keep the state as objects on heap, reusing mutable objects
> > can theoretically create cases where the same object is used in multiple
> > state mappings.
> >
> > The effect is that many people that try Flink get much lower performance
> > than they could possibly get. From empirical evidence, almost all users
> > that I (Stephan) have been in touch with eventually run into this issue
> > eventually.
> >
> > There are multiple observations about that design:
> >
> >
> > -
> >
> > Object copies are extremely costly. While some simple copy virtually for
> > free (types reliably detected as immutable are not copied at all), many
> > real pipelines use types like Avro, Thrift, JSON, etc, which are very
> > expensive to copy.
> >
> >
> >
> > -
> >
> > Keyed operations currently only occur after shuffles. The operations are
> > hence the first in a pipeline and will never have a reused object anyways.
> > That means for the most critical operation, this pre-caution is
> > unnecessary.
> >
> >
> >
> > -
> >
> > The mode is inconsistent with the contract of the DataSet API, which
> > does not copy at each step
> >
> >
> >
> > -
> >
> > To prevent these copies, users can select {{enableObjectReuse()}}, which
> > is misleading, since it does not really reuse mutable objects, but avoids
> > additional copies.
> >
> >
> > Proposal
> >
> > Summary
> >
> > I propose to change the default behavior of the DataStream runtime to be
> > the same as the DataSet runtime. That means that new objects are chosen on
> > every deserialization, and no copies are made as the objects are passed on
> > along the pipelines.
> >
> > Details
> >
> > I propose to drop the execution config flag {{objectReuse}} and instead
> > introduce an {{ObjectReuseMode}} enumeration with better control of what
> > should happen. There will be three different types:
> >
> >
> > -
> >
> > DEFAULT
> >
[jira] [Created] (FLINK-7021) Flink Task Manager hangs on startup if one Zookeeper node is unresolvable
Scott Kidder created FLINK-7021:
---
Summary: Flink Task Manager hangs on startup if one Zookeeper node
is unresolvable
Key: FLINK-7021
URL: https://issues.apache.org/jira/browse/FLINK-7021
Project: Flink
Issue Type: Bug
Components: Core
Affects Versions: 1.2.1, 1.3.0, 1.2.0
Environment: Kubernetes cluster running:
* Flink 1.3.0 Job Manager & Task Manager on Java 8u131
* Zookeeper 3.4.10 cluster with 3 nodes
Reporter: Scott Kidder
h2. Problem
Flink Task Manager will hang during startup if one of the Zookeeper nodes in
the Zookeeper connection string is unresolvable.
h2. Expected Behavior
Flink should retry name resolution & connection to Zookeeper nodes with
exponential back-off.
h2. Environment Details
We're running Flink and Zookeeper in Kubernetes on CoreOS. CoreOS can run in a
configuration that automatically detects and applies operating system updates.
We have a Zookeeper node running on the same CoreOS instance as Flink. It's
possible that the Zookeeper node will not yet be started when the Flink
components are started. This could cause hostname resolution of the Zookeeper
nodes to fail.
h3. Flink Task Manager Logs
{noformat}
2017-06-27 15:38:51,713 INFO org.apache.flink.runtime.taskmanager.TaskManager
- Using configured hostname/address for TaskManager: 10.2.45.11
2017-06-27 15:38:51,714 INFO org.apache.flink.runtime.taskmanager.TaskManager
- Starting TaskManager
2017-06-27 15:38:51,714 INFO org.apache.flink.runtime.taskmanager.TaskManager
- Starting TaskManager actor system at 10.2.45.11:6122.
2017-06-27 15:38:52,950 INFO akka.event.slf4j.Slf4jLogger
- Slf4jLogger started
2017-06-27 15:38:53,079 INFO Remoting
- Starting remoting
2017-06-27 15:38:53,573 INFO Remoting
- Remoting started; listening on addresses
:[akka.tcp://flink@10.2.45.11:6122]
2017-06-27 15:38:53,576 INFO org.apache.flink.runtime.taskmanager.TaskManager
- Starting TaskManager actor
2017-06-27 15:38:53,660 INFO
org.apache.flink.runtime.io.network.netty.NettyConfig - NettyConfig
[server address: /10.2.45.11, server port: 6121, ssl enabled: false, memory
segment size (bytes): 32768, transport type: NIO, number of server threads: 2
(manual), number of client threads: 2 (manual), server connect backlog: 0 (use
Netty's default), client connect timeout (sec): 120, send/receive buffer size
(bytes): 0 (use Netty's default)]
2017-06-27 15:38:53,682 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerConfiguration - Messages have
a max timeout of 1 ms
2017-06-27 15:38:53,688 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerServices - Temporary file
directory '/tmp': total 49 GB, usable 42 GB (85.71% usable)
2017-06-27 15:38:54,071 INFO
org.apache.flink.runtime.io.network.buffer.NetworkBufferPool - Allocated 96 MB
for network buffer pool (number of memory segments: 3095, bytes per segment:
32768).
2017-06-27 15:38:54,564 INFO
org.apache.flink.runtime.io.network.NetworkEnvironment- Starting the
network environment and its components.
2017-06-27 15:38:54,576 INFO
org.apache.flink.runtime.io.network.netty.NettyClient - Successful
initialization (took 4 ms).
2017-06-27 15:38:54,677 INFO
org.apache.flink.runtime.io.network.netty.NettyServer - Successful
initialization (took 101 ms). Listening on SocketAddress /10.2.45.11:6121.
2017-06-27 15:38:54,981 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerServices - Limiting
managed memory to 0.7 of the currently free heap space (612 MB), memory will be
allocated lazily.
2017-06-27 15:38:55,050 INFO
org.apache.flink.runtime.io.disk.iomanager.IOManager - I/O manager
uses directory /tmp/flink-io-ca01554d-f25e-4c17-a828-96d82b43d4a7 for spill
files.
2017-06-27 15:38:55,061 INFO org.apache.flink.runtime.metrics.MetricRegistry
- Configuring StatsDReporter with {interval=10 SECONDS, port=8125,
host=localhost, class=org.apache.flink.metrics.statsd.StatsDReporter}.
2017-06-27 15:38:55,065 INFO org.apache.flink.metrics.statsd.StatsDReporter
- Configured StatsDReporter with {host:localhost, port:8125}
2017-06-27 15:38:55,065 INFO org.apache.flink.runtime.metrics.MetricRegistry
- Periodically reporting metrics in intervals of 10 SECONDS for
reporter statsd of type org.apache.flink.metrics.statsd.StatsDReporter.
2017-06-27 15:38:55,175 INFO org.apache.flink.runtime.filecache.FileCache
- User file cache uses directory
/tmp/flink-dist-cache-e4c5bcc5-7513-40d9-a665-0d33c80a36ba
2017-06-27 15:38:55,187 INFO org.apache.flink.runtime.filecache.FileCache
- User file cache uses director
[jira] [Created] (FLINK-7022) Flink Job Manager Scheduler & Web Frontend out of sync when Zookeeper is unavailable on startup
Scott Kidder created FLINK-7022:
---
Summary: Flink Job Manager Scheduler & Web Frontend out of sync
when Zookeeper is unavailable on startup
Key: FLINK-7022
URL: https://issues.apache.org/jira/browse/FLINK-7022
Project: Flink
Issue Type: Bug
Components: JobManager
Affects Versions: 1.2.1, 1.3.0, 1.2.0
Environment: Kubernetes cluster running:
* Flink 1.3.0 Job Manager & Task Manager on Java 8u131
* Zookeeper 3.4.10 cluster with 3 nodes
Reporter: Scott Kidder
h2. Problem
Flink Job Manager web frontend is permanently unavailable if one or more
Zookeeper nodes are unresolvable during startup. The job scheduler eventually
recovers and assigns jobs to task managers, but the web frontend continues to
respond with an HTTP 503 and the following message:
{noformat}Service temporarily unavailable due to an ongoing leader election.
Please refresh.{noformat}
h2. Expected Behavior
Once Flink is able to interact with Zookeeper successfully, all aspects of the
Job Manager (job scheduling & the web frontend) should be available.
h2. Environment Details
We're running Flink and Zookeeper in Kubernetes on CoreOS. CoreOS can run in a
configuration that automatically detects and applies operating system updates.
We have a Zookeeper node running on the same CoreOS instance as Flink. It's
possible that the Zookeeper node will not yet be started when the Flink
components are started. This could cause hostname resolution of the Zookeeper
nodes to fail.
h3. Flink Task Manager Logs
{noformat}
2017-06-27 15:38:47,161 INFO
org.apache.flink.configuration.GlobalConfiguration- Loading
configuration property: metrics.reporter.statsd.host, localhost
2017-06-27 15:38:47,161 INFO
org.apache.flink.configuration.GlobalConfiguration- Loading
configuration property: metrics.reporter.statsd.port, 8125
2017-06-27 15:38:47,162 INFO
org.apache.flink.configuration.GlobalConfiguration- Loading
configuration property: metrics.reporter.statsd.interval, 10 SECONDS
2017-06-27 15:38:47,254 INFO
org.apache.flink.configuration.GlobalConfiguration- Loading
configuration property: state.backend, filesystem
2017-06-27 15:38:47,254 INFO
org.apache.flink.configuration.GlobalConfiguration- Loading
configuration property: state.backend.fs.checkpointdir,
hdfs://hdfs:8020/flink/checkpoints
2017-06-27 15:38:47,255 INFO
org.apache.flink.configuration.GlobalConfiguration- Loading
configuration property: state.savepoints.dir, hdfs://hdfs:8020/flink/savepoints
2017-06-27 15:38:47,255 INFO
org.apache.flink.configuration.GlobalConfiguration- Loading
configuration property: recovery.mode, zookeeper
2017-06-27 15:38:47,256 INFO
org.apache.flink.configuration.GlobalConfiguration- Loading
configuration property: recovery.zookeeper.quorum,
zookeeper-0.zookeeper:2181,zookeeper-1.zookeeper:2181,zookeeper-2.zookeeper:2181
2017-06-27 15:38:47,256 INFO
org.apache.flink.configuration.GlobalConfiguration- Loading
configuration property: recovery.zookeeper.storageDir,
hdfs://hdfs:8020/flink/recovery
2017-06-27 15:38:47,256 INFO
org.apache.flink.configuration.GlobalConfiguration- Loading
configuration property: recovery.jobmanager.port, 6123
2017-06-27 15:38:47,257 INFO
org.apache.flink.configuration.GlobalConfiguration- Loading
configuration property: blob.server.port, 41479
2017-06-27 15:38:47,357 WARN org.apache.flink.configuration.Configuration
- Config uses deprecated configuration key 'recovery.mode' instead
of proper key 'high-availability'
2017-06-27 15:38:47,366 INFO org.apache.flink.runtime.jobmanager.JobManager
- Starting JobManager with high-availability
2017-06-27 15:38:47,366 WARN org.apache.flink.configuration.Configuration
- Config uses deprecated configuration key
'recovery.jobmanager.port' instead of proper key
'high-availability.jobmanager.port'
2017-06-27 15:38:47,452 INFO org.apache.flink.runtime.jobmanager.JobManager
- Starting JobManager on flink:6123 with execution mode CLUSTER
2017-06-27 15:38:47,549 INFO
org.apache.flink.configuration.GlobalConfiguration- Loading
configuration property: jobmanager.rpc.address, flink
2017-06-27 15:38:47,549 INFO
org.apache.flink.configuration.GlobalConfiguration- Loading
configuration property: jobmanager.rpc.port, 6123
2017-06-27 15:38:47,549 INFO
org.apache.flink.configuration.GlobalConfiguration- Loading
configuration property: jobmanager.heap.mb, 1024
2017-06-27 15:38:47,549 INFO
org.apache.flink.configuration.GlobalConfiguration- Loading
configuration property: taskmanager.heap.mb, 1024
2017-06-27 15:38:47,549 INFO
org.apache.flink.configuration.GlobalConfiguration
[jira] [Created] (FLINK-7023) Remaining types for Gelly ValueArrays
Greg Hogan created FLINK-7023: - Summary: Remaining types for Gelly ValueArrays Key: FLINK-7023 URL: https://issues.apache.org/jira/browse/FLINK-7023 Project: Flink Issue Type: Sub-task Components: Gelly Affects Versions: 1.4.0 Reporter: Greg Hogan Assignee: Greg Hogan Priority: Trivial Fix For: 1.4.0 Add implementations of Byte/Char/Double/Float/ShortValueArray. Along with the existing implementations of Int/Long/Null/StringValueArray this covers all 10 CopyableValue types. -- This message was sent by Atlassian JIRA (v6.4.14#64029)
[jira] [Created] (FLINK-7024) Add supported for selecting window proctime/rowtime on row-based Tumble/Slide window
sunjincheng created FLINK-7024:
--
Summary: Add supported for selecting window proctime/rowtime on
row-based Tumble/Slide window
Key: FLINK-7024
URL: https://issues.apache.org/jira/browse/FLINK-7024
Project: Flink
Issue Type: Bug
Components: Table API & SQL
Affects Versions: 1.4.0
Reporter: sunjincheng
Assignee: sunjincheng
We get validate exception,when selecting window.proctime/rowtime on row-based
group window.
{code}
table
.window(Tumble over 2.rows on 'proctime as 'w)
.groupBy('w, 'string)
.select('string, countFun('string) as 'cnt, 'w.rowtime as 'proctime)
.window(Over partitionBy 'string orderBy 'proctime preceding
UNBOUNDED_RANGE following CURRENT_RANGE as 'w2)
.select('string, 'cnt.sum over 'w2 as 'cnt)
{code}
Exception:
{code}
org.apache.flink.table.api.ValidationException: Window start and Window end
cannot be selected for a row-count Tumbling window.
at
org.apache.flink.table.plan.logical.LogicalNode.failValidation(LogicalNode.scala:143)
at
org.apache.flink.table.plan.logical.WindowAggregate.validate(operators.scala:660)
{code}
We should add window.proctime/rowtime check in `validate ` method.
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
[jira] [Created] (FLINK-7025) Using NullByteKeySelector for Unbounded ProcTime NonPartitioned Over
sunjincheng created FLINK-7025: -- Summary: Using NullByteKeySelector for Unbounded ProcTime NonPartitioned Over Key: FLINK-7025 URL: https://issues.apache.org/jira/browse/FLINK-7025 Project: Flink Issue Type: Bug Reporter: sunjincheng Assignee: sunjincheng Currently we added `Cleanup State` feature. But It not work well if we enabled the stateCleaning on Unbounded ProcTime NonPartitioned Over window, Because in `ProcessFunctionWithCleanupState` we has using the keyed state. So, In this JIRA. I'll change the `Unbounded ProcTime NonPartitioned Over` to `partitioned Over` by using NullByteKeySelector. OR created a `NonKeyedProcessFunctionWithCleanupState`. But I think the first way is simpler. What do you think? [~fhueske] -- This message was sent by Atlassian JIRA (v6.4.14#64029)
