[jira] [Commented] (FLINK-10373) CoGroup Window support lateness
[ https://issues.apache.org/jira/browse/FLINK-10373?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16621519#comment-16621519 ] Kostas Kloudas commented on FLINK-10373: Hi [~zhenhua]. Isn't this the same as https://issues.apache.org/jira/browse/FLINK-10050 ? > CoGroup Window support lateness > > > Key: FLINK-10373 > URL: https://issues.apache.org/jira/browse/FLINK-10373 > Project: Flink > Issue Type: Improvement > Components: DataStream API >Reporter: huangzhenhua >Priority: Critical > Labels: cogroup, lateness > > I suggest that CoGroup Window has the lateness event process > function。Inside my business, we hope that the window created by two > stream join can process lateness event. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (FLINK-9924) Upgrade zookeeper to 3.4.13
[ https://issues.apache.org/jira/browse/FLINK-9924?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ted Yu updated FLINK-9924: -- Description: zookeeper 3.4.13 is being released. ZOOKEEPER-2959 fixes data loss when observer is used ZOOKEEPER-2184 allows ZooKeeper Java clients to work in dynamic IP (container / cloud) environment was: zookeeper 3.4.13 is being released. ZOOKEEPER-2959 fixes data loss when observer is used ZOOKEEPER-2184 allows ZooKeeper Java clients to work in dynamic IP (container / cloud) environment > Upgrade zookeeper to 3.4.13 > --- > > Key: FLINK-9924 > URL: https://issues.apache.org/jira/browse/FLINK-9924 > Project: Flink > Issue Type: Task >Reporter: Ted Yu >Assignee: vinoyang >Priority: Major > > zookeeper 3.4.13 is being released. > ZOOKEEPER-2959 fixes data loss when observer is used > ZOOKEEPER-2184 allows ZooKeeper Java clients to work in dynamic IP (container > / cloud) environment -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-10373) CoGroup Window support lateness
huangzhenhua created FLINK-10373: Summary: CoGroup Window support lateness Key: FLINK-10373 URL: https://issues.apache.org/jira/browse/FLINK-10373 Project: Flink Issue Type: Improvement Components: DataStream API Reporter: huangzhenhua I suggest that CoGroup Window has the lateness event process function。Inside my business, we hope that the window created by two stream join can process lateness event. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10245) Add DataStream HBase Sink
[ https://issues.apache.org/jira/browse/FLINK-10245?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16621468#comment-16621468 ] Shimin Yang commented on FLINK-10245: - Hi [~hequn8128], I have updated the document, we can discuss how to implement the merge in there. > Add DataStream HBase Sink > - > > Key: FLINK-10245 > URL: https://issues.apache.org/jira/browse/FLINK-10245 > Project: Flink > Issue Type: Sub-task > Components: Streaming Connectors >Reporter: Shimin Yang >Assignee: Shimin Yang >Priority: Major > Labels: pull-request-available > > Design documentation: > [https://docs.google.com/document/d/1of0cYd73CtKGPt-UL3WVFTTBsVEre-TNRzoAt5u2PdQ/edit?usp=sharing] -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-10372) There is no API to configure the timer state backend
Elias Levy created FLINK-10372:
--
Summary: There is no API to configure the timer state backend
Key: FLINK-10372
URL: https://issues.apache.org/jira/browse/FLINK-10372
Project: Flink
Issue Type: Improvement
Components: DataStream API, State Backends, Checkpointing
Affects Versions: 1.6.0
Reporter: Elias Levy
Flink 1.6.0, via FLINK-9485, introduced the option to store timers in RocksDB
instead of the heap. Alas, this can only be configured via the
{{state.backend.rocksdb.timer-service.factory}} config file option. That means
that the choice of state backend to use for timer can't be made on a per job
basis on a shared cluster.
There is a need for an API in {{RocksDBStateBackend}} to configure the backend
per job.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-8424) [Cassandra Connector] Update Cassandra version to last one
[ https://issues.apache.org/jira/browse/FLINK-8424?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16621229#comment-16621229 ] ASF GitHub Bot commented on FLINK-8424: --- bmeriaux commented on a change in pull request #6715: FLINK-8424: update cassandra and driver version to latest URL: https://github.com/apache/flink/pull/6715#discussion_r218969992 ## File path: flink-connectors/flink-connector-cassandra/pom.xml ## @@ -37,8 +37,8 @@ under the License. - 2.2.5 - 3.0.0 + 3.11.3 Review comment: done :) This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > [Cassandra Connector] Update Cassandra version to last one > -- > > Key: FLINK-8424 > URL: https://issues.apache.org/jira/browse/FLINK-8424 > Project: Flink > Issue Type: Improvement > Components: Cassandra Connector >Reporter: Joao Boto >Priority: Major > Labels: pull-request-available > > Cassandra connector are using a version outdated > This is to upgrade the cassandra version to something new > https://git1-us-west.apache.org/repos/asf?p=cassandra.git;a=blob_plain;f=CHANGES.txt;hb=refs/tags/cassandra-3.11.1 -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[GitHub] bmeriaux commented on a change in pull request #6715: FLINK-8424: update cassandra and driver version to latest
bmeriaux commented on a change in pull request #6715: FLINK-8424: update cassandra and driver version to latest URL: https://github.com/apache/flink/pull/6715#discussion_r218969992 ## File path: flink-connectors/flink-connector-cassandra/pom.xml ## @@ -37,8 +37,8 @@ under the License. - 2.2.5 - 3.0.0 + 3.11.3 Review comment: done :) This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Comment Edited] (FLINK-10079) Automatically register sink table from external catalogs
[
https://issues.apache.org/jira/browse/FLINK-10079?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16619847#comment-16619847
]
Fabian Hueske edited comment on FLINK-10079 at 9/19/18 8:01 PM:
Fixed for 1.6.2 with 65cd7b55eb7c0c7af840eebdddf83ae46d28e204
Fixed for 1.7.0 with cc37d7a7555aa7f18da2d0eb62c569cb080332ed
was (Author: fhueske):
Fixed for 1.6.2 with 57b621c8c1b39cbc42cca49212ba2b9bb7ef1a2b
Fixed for 1.7.0 with cc37d7a7555aa7f18da2d0eb62c569cb080332ed
> Automatically register sink table from external catalogs
>
>
> Key: FLINK-10079
> URL: https://issues.apache.org/jira/browse/FLINK-10079
> Project: Flink
> Issue Type: Improvement
> Components: Table API SQL
>Affects Versions: 1.6.0
>Reporter: Jun Zhang
>Assignee: Jun Zhang
>Priority: Major
> Fix For: 1.7.0, 1.6.2
>
>
> In the documentation, there is a description:
> {quote}Once registered in a TableEnvironment, all tables defined in a
> ExternalCatalog can be accessed from Table API or SQL queries by specifying
> their full path, such as catalog.database.table.
> {quote}
> Currently, this is true only for source tables. For sink table (specified in
> the Table API or SQL), the users have to explicitly register it even though
> it is defined in a registered ExternalCatalog, otherwise "No table was
> registered under the name XXX" TableException would be thrown.
> It would be better keep consistent between source table and sink table, and
> the users would enjoy more convenient approach to inserting into sink tables.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Reopened] (FLINK-10079) Automatically register sink table from external catalogs
[
https://issues.apache.org/jira/browse/FLINK-10079?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Fabian Hueske reopened FLINK-10079:
---
> Automatically register sink table from external catalogs
>
>
> Key: FLINK-10079
> URL: https://issues.apache.org/jira/browse/FLINK-10079
> Project: Flink
> Issue Type: Improvement
> Components: Table API SQL
>Affects Versions: 1.6.0
>Reporter: Jun Zhang
>Assignee: Jun Zhang
>Priority: Major
> Fix For: 1.7.0, 1.6.2
>
>
> In the documentation, there is a description:
> {quote}Once registered in a TableEnvironment, all tables defined in a
> ExternalCatalog can be accessed from Table API or SQL queries by specifying
> their full path, such as catalog.database.table.
> {quote}
> Currently, this is true only for source tables. For sink table (specified in
> the Table API or SQL), the users have to explicitly register it even though
> it is defined in a registered ExternalCatalog, otherwise "No table was
> registered under the name XXX" TableException would be thrown.
> It would be better keep consistent between source table and sink table, and
> the users would enjoy more convenient approach to inserting into sink tables.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Comment Edited] (FLINK-10259) Key validation for GroupWindowAggregate is broken
[
https://issues.apache.org/jira/browse/FLINK-10259?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16619867#comment-16619867
]
Fabian Hueske edited comment on FLINK-10259 at 9/19/18 8:00 PM:
Fixed for 1.5.5 with 5852a25d8d9a4defa70c31ae61231471dc484bc1
Fixed for 1.6.2 with 290d96e50f864468b6b2d2a3b741e0fbdaee5df0
Fixed for 1.7.0 with f28b82909c3c6bcbe0436cae41af9a3c001f1c36
was (Author: fhueske):
Fixed for 1.5.5 with 5852a25d8d9a4defa70c31ae61231471dc484bc1
Fixed for 1.6.2 with 1ee6bc3b99d8f50fb8ed0be0b4936c83ce03fff1
Fixed for 1.7.0 with f28b82909c3c6bcbe0436cae41af9a3c001f1c36
> Key validation for GroupWindowAggregate is broken
> -
>
> Key: FLINK-10259
> URL: https://issues.apache.org/jira/browse/FLINK-10259
> Project: Flink
> Issue Type: Bug
> Components: Table API SQL
>Reporter: Fabian Hueske
>Assignee: Fabian Hueske
>Priority: Major
> Labels: pull-request-available
> Fix For: 1.7.0, 1.6.2, 1.5.5
>
>
> WindowGroups have multiple equivalent keys (start, end) that should be
> handled differently from other keys. The {{UpdatingPlanChecker}} uses
> equivalence groups to identify equivalent keys but the keys of WindowGroups
> are not correctly assigned to groups.
> This means that we cannot correctly extract keys from queries that use group
> windows.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Reopened] (FLINK-10259) Key validation for GroupWindowAggregate is broken
[
https://issues.apache.org/jira/browse/FLINK-10259?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Fabian Hueske reopened FLINK-10259:
---
> Key validation for GroupWindowAggregate is broken
> -
>
> Key: FLINK-10259
> URL: https://issues.apache.org/jira/browse/FLINK-10259
> Project: Flink
> Issue Type: Bug
> Components: Table API SQL
>Reporter: Fabian Hueske
>Assignee: Fabian Hueske
>Priority: Major
> Labels: pull-request-available
> Fix For: 1.7.0, 1.6.2, 1.5.5
>
>
> WindowGroups have multiple equivalent keys (start, end) that should be
> handled differently from other keys. The {{UpdatingPlanChecker}} uses
> equivalence groups to identify equivalent keys but the keys of WindowGroups
> are not correctly assigned to groups.
> This means that we cannot correctly extract keys from queries that use group
> windows.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Resolved] (FLINK-10332) Move data available notification in PipelinedSubpartition out of the synchronized block
[
https://issues.apache.org/jira/browse/FLINK-10332?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Nico Kruber resolved FLINK-10332.
-
Resolution: Fixed
Fix Version/s: 1.5.5
1.6.2
1.7.0
Fixed via:
- release-1.5: 607c5675a108114254c747ca3565c6f03a98434e
- release-1.6: 5b2b0f71cb33fa749b9332ffd386f39c9acf6a0e
- master: c15ba1cd5a97ced3bb6411d588c3fb68df8a2869
> Move data available notification in PipelinedSubpartition out of the
> synchronized block
> ---
>
> Key: FLINK-10332
> URL: https://issues.apache.org/jira/browse/FLINK-10332
> Project: Flink
> Issue Type: Sub-task
> Components: Network
>Affects Versions: 1.5.0, 1.5.1, 1.5.2, 1.5.3, 1.6.0, 1.7.0
>Reporter: Nico Kruber
>Assignee: Nico Kruber
>Priority: Major
> Labels: pull-request-available
> Fix For: 1.7.0, 1.6.2, 1.5.5
>
>
> Currently, calls to {{PipelinedSubpartition#notifyDataAvailable();}} are
> unnecessarily executed inside a {{synchronized (buffers)}} block which may
> lead to lock contention.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Resolved] (FLINK-10331) Fix unnecessary flush requests to the network stack
[ https://issues.apache.org/jira/browse/FLINK-10331?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Nico Kruber resolved FLINK-10331. - Resolution: Fixed Fix Version/s: 1.5.5 1.6.2 1.7.0 Fixed via: - release-1.5: 16ea186fef3ac7eee8679f8e87132313508573b3 - release-1.6: ef192266e94c088b05d611e25165b7efdb0585d8 - master: a9e5f70515a3cbad7e776b9248e841f6c81c9b7a > Fix unnecessary flush requests to the network stack > --- > > Key: FLINK-10331 > URL: https://issues.apache.org/jira/browse/FLINK-10331 > Project: Flink > Issue Type: Sub-task > Components: Network >Affects Versions: 1.5.0, 1.5.1, 1.5.2, 1.5.3, 1.6.0, 1.7.0 >Reporter: Nico Kruber >Assignee: Nico Kruber >Priority: Major > Labels: pull-request-available > Fix For: 1.7.0, 1.6.2, 1.5.5 > > > With the re-design of the record writer interaction with the > result(sub)partitions, flush requests can currently pile up in these > scenarios: > - a previous flush request has not been completely handled yet and/or is > still enqueued or > - the network stack is still polling from this subpartition and doesn't need > a new notification > These lead to increased notifications in low latency settings (low output > flusher intervals) which can be avoided. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[GitHub] TisonKun commented on issue #6716: [hotfix] [yarn-test] Clean up inactive test
TisonKun commented on issue #6716: [hotfix] [yarn-test] Clean up inactive test URL: https://github.com/apache/flink/pull/6716#issuecomment-422914196 FYI, Travis fails on legacy test `JobManagerFailsTest`. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] satybald commented on a change in pull request #6706: [hotfix][doc] Fix curl example in upload jar example
satybald commented on a change in pull request #6706: [hotfix][doc] Fix curl
example in upload jar example
URL: https://github.com/apache/flink/pull/6706#discussion_r218899758
##
File path:
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarUploadHeaders.java
##
@@ -73,7 +73,7 @@ public static JarUploadHeaders getInstance() {
public String getDescription() {
return "Uploads a jar to the cluster. The jar must be sent as
multi-part data. Make sure that the \"Content-Type\"" +
" header is set to \"application/x-java-archive\", as
some http libraries do not add the header by default.\n" +
- "Using 'curl' you can upload a jar via 'curl -X POST -H
\"Expect:\" -F \"jarfile=#path/to/flink-job.jar\" http://hostname:port; + URL +
"'.";
+ "Using 'curl' you can upload a jar via 'curl -X POST -F
\"jarfile=@path/to/flink-job.jar;type=application/x-java-archive\"
http://hostname:port; + URL + "'.";
Review comment:
@GJL thank you for the feedback. agree, it's Content-Type comment is quite
confusing. Indeed Flink(1.6.0) doesn't need to set `content-type`, curl works
without it.
I'll update curl example to substitute # => @ symbols in this case.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[jira] [Commented] (FLINK-10134) UTF-16 support for TextInputFormat
[
https://issues.apache.org/jira/browse/FLINK-10134?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620859#comment-16620859
]
ASF GitHub Bot commented on FLINK-10134:
XuQianJin-Stars commented on issue #6710: [FLINK-10134] UTF-16 support for
TextInputFormat bug fixed
URL: https://github.com/apache/flink/pull/6710#issuecomment-422881621
> Hi @XuQianJin-Stars , thanks for your PR. Could you rebase your commits?
@Clark Thank you. I have re submitted PR. You can help me check my PR
and give you advice.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> UTF-16 support for TextInputFormat
> --
>
> Key: FLINK-10134
> URL: https://issues.apache.org/jira/browse/FLINK-10134
> Project: Flink
> Issue Type: Bug
> Components: Core
>Affects Versions: 1.4.2
>Reporter: David Dreyfus
>Priority: Blocker
> Labels: pull-request-available
>
> It does not appear that Flink supports a charset encoding of "UTF-16". It
> particular, it doesn't appear that Flink consumes the Byte Order Mark (BOM)
> to establish whether a UTF-16 file is UTF-16LE or UTF-16BE.
>
> TextInputFormat.setCharset("UTF-16") calls DelimitedInputFormat.setCharset(),
> which sets TextInputFormat.charsetName and then modifies the previously set
> delimiterString to construct the proper byte string encoding of the the
> delimiter. This same charsetName is also used in TextInputFormat.readRecord()
> to interpret the bytes read from the file.
>
> There are two problems that this implementation would seem to have when using
> UTF-16.
> # delimiterString.getBytes(getCharset()) in DelimitedInputFormat.java will
> return a Big Endian byte sequence including the Byte Order Mark (BOM). The
> actual text file will not contain a BOM at each line ending, so the delimiter
> will never be read. Moreover, if the actual byte encoding of the file is
> Little Endian, the bytes will be interpreted incorrectly.
> # TextInputFormat.readRecord() will not see a BOM each time it decodes a
> byte sequence with the String(bytes, offset, numBytes, charset) call.
> Therefore, it will assume Big Endian, which may not always be correct. [1]
> [https://github.com/apache/flink/blob/master/flink-java/src/main/java/org/apache/flink/api/java/io/TextInputFormat.java#L95]
>
> While there are likely many solutions, I would think that all of them would
> have to start by reading the BOM from the file when a Split is opened and
> then using that BOM to modify the specified encoding to a BOM specific one
> when the caller doesn't specify one, and to overwrite the caller's
> specification if the BOM is in conflict with the caller's specification. That
> is, if the BOM indicates Little Endian and the caller indicates UTF-16BE,
> Flink should rewrite the charsetName as UTF-16LE.
> I hope this makes sense and that I haven't been testing incorrectly or
> misreading the code.
>
> I've verified the problem on version 1.4.2. I believe the problem exists on
> all versions.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[GitHub] XuQianJin-Stars commented on issue #6710: [FLINK-10134] UTF-16 support for TextInputFormat bug fixed
XuQianJin-Stars commented on issue #6710: [FLINK-10134] UTF-16 support for TextInputFormat bug fixed URL: https://github.com/apache/flink/pull/6710#issuecomment-422881621 > Hi @XuQianJin-Stars , thanks for your PR. Could you rebase your commits? @Clark Thank you. I have re submitted PR. You can help me check my PR and give you advice. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-10358) Flink kinesis connector could throw NPE during getRecords() call
[ https://issues.apache.org/jira/browse/FLINK-10358?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620832#comment-16620832 ] ASF GitHub Bot commented on FLINK-10358: yxu-valleytider commented on issue #6708: [FLINK-10358][Kinesis-connector] fix NPE in case millisBehindLatest is null URL: https://github.com/apache/flink/pull/6708#issuecomment-422874733 @mxm We do plan to contribute the efforts of consuming from dynamodb streams (Flink-4582) back to upstream flink. This patch is needed for our internal testing. And we strive to keep upstream flink and our internal fork as consistent as possible. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Flink kinesis connector could throw NPE during getRecords() call > - > > Key: FLINK-10358 > URL: https://issues.apache.org/jira/browse/FLINK-10358 > Project: Flink > Issue Type: Bug > Components: Kinesis Connector >Reporter: Ying Xu >Assignee: Ying Xu >Priority: Major > Labels: pull-request-available > > When extending the flink kinesis connector to consume from a dynamodb stream, > it was found NPE could be thrown at > [here|https://github.com/apache/flink/blob/e3c98f05d3544d0165c2d97d2d00fcd295cef8c8/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/ShardConsumer.java#L376] > . > This is because the [getRecords > API|https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_streams_GetRecords.html] > in dynamodb streams does not return the millisBehindLatest field and has it > set to null. Null check is probably needed here. > See FLINK-4582 for the context of building dynamodb streams connector on top > of the Kinesis connector. > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[GitHub] yxu-valleytider commented on issue #6708: [FLINK-10358][Kinesis-connector] fix NPE in case millisBehindLatest is null
yxu-valleytider commented on issue #6708: [FLINK-10358][Kinesis-connector] fix NPE in case millisBehindLatest is null URL: https://github.com/apache/flink/pull/6708#issuecomment-422874733 @mxm We do plan to contribute the efforts of consuming from dynamodb streams (Flink-4582) back to upstream flink. This patch is needed for our internal testing. And we strive to keep upstream flink and our internal fork as consistent as possible. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-10265) Configure checkpointing behavior for SQL Client
[
https://issues.apache.org/jira/browse/FLINK-10265?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620773#comment-16620773
]
vinoyang commented on FLINK-10265:
--
[~twalthr] any opinion and suggestion?
> Configure checkpointing behavior for SQL Client
> ---
>
> Key: FLINK-10265
> URL: https://issues.apache.org/jira/browse/FLINK-10265
> Project: Flink
> Issue Type: New Feature
> Components: SQL Client
>Reporter: Timo Walther
>Assignee: vinoyang
>Priority: Major
>
> The SQL Client environment file should expose checkpointing related
> properties:
> - enable checkpointing
> - checkpointing interval
> - mode
> - timeout
> - etc. see {{org.apache.flink.streaming.api.environment.CheckpointConfig}}
> Per-job selection of state backends and their configuration.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9891) Flink cluster is not shutdown in YARN mode when Flink client is stopped
[
https://issues.apache.org/jira/browse/FLINK-9891?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620770#comment-16620770
]
ASF GitHub Bot commented on FLINK-9891:
---
GJL commented on a change in pull request #6718: [FLINK-9891] Add optional hook
to shutdown cluster if a session was created in per-job mode in attached mode
URL: https://github.com/apache/flink/pull/6718#discussion_r218859953
##
File path: docs/ops/cli.md
##
@@ -265,6 +268,9 @@ Action "run" compiles and runs a program.
connect to a different JobManager
than
the one specified in the
configuration.
+ -sae,--schutdownOnAttachedExist If the job mode is not detached and
Review comment:
Is this supposed to be _exit_ instead of _exist_? How about
`--shutdownOnExit`?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Flink cluster is not shutdown in YARN mode when Flink client is stopped
> ---
>
> Key: FLINK-9891
> URL: https://issues.apache.org/jira/browse/FLINK-9891
> Project: Flink
> Issue Type: Bug
>Affects Versions: 1.5.0, 1.5.1
>Reporter: Sergey Krasovskiy
>Assignee: Shuyi Chen
>Priority: Major
> Labels: pull-request-available
>
> We are not using session mode and detached mode. The command to run Flink job
> on YARN is:
> {code:java}
> /bin/flink run -m yarn-cluster -yn 1 -yqu flink -yjm 768 -ytm
> 2048 -j ./flink-quickstart-java-1.0-SNAPSHOT.jar -c org.test.WordCount
> {code}
> Flink CLI logs:
> {code:java}
> Setting HADOOP_CONF_DIR=/etc/hadoop/conf because no HADOOP_CONF_DIR was set.
> SLF4J: Class path contains multiple SLF4J bindings.
> SLF4J: Found binding in
> [jar:file:/opt/flink-streaming/flink-streaming-1.5.1-1.5.1-bin-hadoop27-scala_2.11-1531485329/lib/slf4j-log4j12-1.7.7.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> SLF4J: Found binding in
> [jar:file:/usr/hdp/2.4.2.10-1/hadoop/lib/slf4j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an
> explanation.
> SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
> 2018-07-18 12:47:03,747 INFO
> org.apache.hadoop.yarn.client.api.impl.TimelineClientImpl - Timeline service
> address: http://hmaster-1.ipbl.rgcloud.net:8188/ws/v1/timeline/
> 2018-07-18 12:47:04,222 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli -
> No path for the flink jar passed. Using the location of class
> org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
> 2018-07-18 12:47:04,222 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli -
> No path for the flink jar passed. Using the location of class
> org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
> 2018-07-18 12:47:04,248 WARN
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Neither the
> HADOOP_CONF_DIR nor the YARN_CONF_DIR environment variable is set. The Flink
> YARN Client needs one of these to be set to properly load the Hadoop
> configuration for accessing YARN.
> 2018-07-18 12:47:04,409 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Cluster specification:
> ClusterSpecification{masterMemoryMB=768, taskManagerMemoryMB=2048,
> numberTaskManagers=1, slotsPerTaskManager=1}
> 2018-07-18 12:47:04,783 WARN
> org.apache.hadoop.hdfs.shortcircuit.DomainSocketFactory - The short-circuit
> local reads feature cannot be used because libhadoop cannot be loaded.
> 2018-07-18 12:47:04,788 WARN
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - The configuration
> directory
> ('/opt/flink-streaming/flink-streaming-1.5.1-1.5.1-bin-hadoop27-scala_2.11-1531485329/conf')
> contains both LOG4J and Logback configuration files. Please delete or rename
> one of them.
> 2018-07-18 12:47:07,846 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Submitting application
> master application_1531474158783_10814
> 2018-07-18 12:47:08,073 INFO
> org.apache.hadoop.yarn.client.api.impl.YarnClientImpl - Submitted application
> application_1531474158783_10814
> 2018-07-18 12:47:08,074 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Waiting for the cluster
> to be allocated
> 2018-07-18 12:47:08,076 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Deploying cluster,
> current state ACCEPTED
> 2018-07-18 12:47:12,864 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - YARN application has
> been
[GitHub] GJL commented on a change in pull request #6718: [FLINK-9891] Add optional hook to shutdown cluster if a session was created in per-job mode in attached mode
GJL commented on a change in pull request #6718: [FLINK-9891] Add optional hook to shutdown cluster if a session was created in per-job mode in attached mode URL: https://github.com/apache/flink/pull/6718#discussion_r218859953 ## File path: docs/ops/cli.md ## @@ -265,6 +268,9 @@ Action "run" compiles and runs a program. connect to a different JobManager than the one specified in the configuration. + -sae,--schutdownOnAttachedExist If the job mode is not detached and Review comment: Is this supposed to be _exit_ instead of _exist_? How about `--shutdownOnExit`? This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-9891) Flink cluster is not shutdown in YARN mode when Flink client is stopped
[
https://issues.apache.org/jira/browse/FLINK-9891?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620758#comment-16620758
]
ASF GitHub Bot commented on FLINK-9891:
---
GJL commented on a change in pull request #6718: [FLINK-9891] Add optional hook
to shutdown cluster if a session was created in per-job mode in attached mode
URL: https://github.com/apache/flink/pull/6718#discussion_r218853732
##
File path:
flink-clients/src/main/java/org/apache/flink/client/cli/ProgramOptions.java
##
@@ -57,6 +58,8 @@
private final boolean detachedMode;
+ private final boolean schutdownOnAttachedExist;
Review comment:
_shutdown_
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Flink cluster is not shutdown in YARN mode when Flink client is stopped
> ---
>
> Key: FLINK-9891
> URL: https://issues.apache.org/jira/browse/FLINK-9891
> Project: Flink
> Issue Type: Bug
>Affects Versions: 1.5.0, 1.5.1
>Reporter: Sergey Krasovskiy
>Assignee: Shuyi Chen
>Priority: Major
> Labels: pull-request-available
>
> We are not using session mode and detached mode. The command to run Flink job
> on YARN is:
> {code:java}
> /bin/flink run -m yarn-cluster -yn 1 -yqu flink -yjm 768 -ytm
> 2048 -j ./flink-quickstart-java-1.0-SNAPSHOT.jar -c org.test.WordCount
> {code}
> Flink CLI logs:
> {code:java}
> Setting HADOOP_CONF_DIR=/etc/hadoop/conf because no HADOOP_CONF_DIR was set.
> SLF4J: Class path contains multiple SLF4J bindings.
> SLF4J: Found binding in
> [jar:file:/opt/flink-streaming/flink-streaming-1.5.1-1.5.1-bin-hadoop27-scala_2.11-1531485329/lib/slf4j-log4j12-1.7.7.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> SLF4J: Found binding in
> [jar:file:/usr/hdp/2.4.2.10-1/hadoop/lib/slf4j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an
> explanation.
> SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
> 2018-07-18 12:47:03,747 INFO
> org.apache.hadoop.yarn.client.api.impl.TimelineClientImpl - Timeline service
> address: http://hmaster-1.ipbl.rgcloud.net:8188/ws/v1/timeline/
> 2018-07-18 12:47:04,222 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli -
> No path for the flink jar passed. Using the location of class
> org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
> 2018-07-18 12:47:04,222 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli -
> No path for the flink jar passed. Using the location of class
> org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
> 2018-07-18 12:47:04,248 WARN
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Neither the
> HADOOP_CONF_DIR nor the YARN_CONF_DIR environment variable is set. The Flink
> YARN Client needs one of these to be set to properly load the Hadoop
> configuration for accessing YARN.
> 2018-07-18 12:47:04,409 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Cluster specification:
> ClusterSpecification{masterMemoryMB=768, taskManagerMemoryMB=2048,
> numberTaskManagers=1, slotsPerTaskManager=1}
> 2018-07-18 12:47:04,783 WARN
> org.apache.hadoop.hdfs.shortcircuit.DomainSocketFactory - The short-circuit
> local reads feature cannot be used because libhadoop cannot be loaded.
> 2018-07-18 12:47:04,788 WARN
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - The configuration
> directory
> ('/opt/flink-streaming/flink-streaming-1.5.1-1.5.1-bin-hadoop27-scala_2.11-1531485329/conf')
> contains both LOG4J and Logback configuration files. Please delete or rename
> one of them.
> 2018-07-18 12:47:07,846 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Submitting application
> master application_1531474158783_10814
> 2018-07-18 12:47:08,073 INFO
> org.apache.hadoop.yarn.client.api.impl.YarnClientImpl - Submitted application
> application_1531474158783_10814
> 2018-07-18 12:47:08,074 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Waiting for the cluster
> to be allocated
> 2018-07-18 12:47:08,076 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Deploying cluster,
> current state ACCEPTED
> 2018-07-18 12:47:12,864 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - YARN application has
> been deployed successfully.
> {code}
> Job Manager logs:
> {code:java}
> 2018-07-18 12:47:09,913 INFO
> org.apache.flink.runtime.entrypoint.ClusterEntrypoint -
>
[jira] [Commented] (FLINK-10157) Allow `null` user values in map state with TTL
[
https://issues.apache.org/jira/browse/FLINK-10157?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620756#comment-16620756
]
ASF GitHub Bot commented on FLINK-10157:
azagrebin commented on a change in pull request #6707: [FLINK-10157] [State
TTL] Allow `null` user values in map state with TTL
URL: https://github.com/apache/flink/pull/6707#discussion_r218853515
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlMapState.java
##
@@ -54,7 +54,13 @@
@Override
public UV get(UK key) throws Exception {
- return getWithTtlCheckAndUpdate(() -> original.get(key), v ->
original.put(key, v), () -> original.remove(key));
+ TtlValue ttlValue = getWrapped(key);
+ return ttlValue == null ? null : ttlValue.getUserValue();
+ }
+
+ private TtlValue getWrapped(UK key) throws Exception {
+ return getWrappedWithTtlCheckAndUpdate(
+ () -> original.get(key), v -> original.put(key, v), ()
-> original.remove(key));
Review comment:
I added it for code reuse because `get` and `contains` needed the same code.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Allow `null` user values in map state with TTL
> --
>
> Key: FLINK-10157
> URL: https://issues.apache.org/jira/browse/FLINK-10157
> Project: Flink
> Issue Type: Bug
> Components: State Backends, Checkpointing
>Affects Versions: 1.6.0
> Environment: Flink:1.6.0
> Scala:2.11
> JDK:1.8
>Reporter: chengjie.wu
>Assignee: Andrey Zagrebin
>Priority: Minor
> Labels: pull-request-available
> Attachments: StateWithOutTtlTest.scala, StateWithTtlTest.scala
>
>
> Thanks for the StateTtl feature,this is exactly what I need now! But I found
> an issue.
> In the previous version or when StateTtl is not enabled,MapState allows
> `null` value,that means after
> {code:java}
> mapState.put("key", null){code}
> , then
> {code:java}
> mapState.contains("key"){code}
> will return {color:#ff}*true*{color}, but when StateTtl is enabled,
> {code:java}
> mapState.contains("key"){code}
> will return {color:#ff}*false*{color}(*the key has not expired*).
> So I think the field `userValue` in
> `org.apache.flink.runtime.state.ttl.TtlValue` should allow `null` value. User
> state is null may not means the TtlValue should be null.
>
> {code:java}
> /**
> * This class wraps user value of state with TTL.
> *
> * @param Type of the user value of state with TTL
> */
> class TtlValue implements Serializable {
> private final T userValue;
> private final long lastAccessTimestamp;
> TtlValue(T userValue, long lastAccessTimestamp) {
> Preconditions.checkNotNull(userValue);
> this.userValue = userValue;
> this.lastAccessTimestamp = lastAccessTimestamp;
> }
> T getUserValue() {
> return userValue;
> }
> long getLastAccessTimestamp() {
> return lastAccessTimestamp;
> }
> }
> {code}
> Am I understanding right?
> This is my test class.
> [^StateWithTtlTest.scala] [^StateWithOutTtlTest.scala]
> ^Thanks!:)^
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9891) Flink cluster is not shutdown in YARN mode when Flink client is stopped
[
https://issues.apache.org/jira/browse/FLINK-9891?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620755#comment-16620755
]
ASF GitHub Bot commented on FLINK-9891:
---
GJL commented on a change in pull request #6718: [FLINK-9891] Add optional hook
to shutdown cluster if a session was created in per-job mode in attached mode
URL: https://github.com/apache/flink/pull/6718#discussion_r218853427
##
File path: docs/ops/cli.md
##
@@ -257,6 +257,9 @@ Action "run" compiles and runs a program.
-s,--fromSavepointPath to a savepoint to restore the
job
from (for example
hdfs:///flink/savepoint-1537).
+ -sae,--schutdownOnAttachedExist If the job mode is not detached and
Review comment:
_shutdown_
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Flink cluster is not shutdown in YARN mode when Flink client is stopped
> ---
>
> Key: FLINK-9891
> URL: https://issues.apache.org/jira/browse/FLINK-9891
> Project: Flink
> Issue Type: Bug
>Affects Versions: 1.5.0, 1.5.1
>Reporter: Sergey Krasovskiy
>Assignee: Shuyi Chen
>Priority: Major
> Labels: pull-request-available
>
> We are not using session mode and detached mode. The command to run Flink job
> on YARN is:
> {code:java}
> /bin/flink run -m yarn-cluster -yn 1 -yqu flink -yjm 768 -ytm
> 2048 -j ./flink-quickstart-java-1.0-SNAPSHOT.jar -c org.test.WordCount
> {code}
> Flink CLI logs:
> {code:java}
> Setting HADOOP_CONF_DIR=/etc/hadoop/conf because no HADOOP_CONF_DIR was set.
> SLF4J: Class path contains multiple SLF4J bindings.
> SLF4J: Found binding in
> [jar:file:/opt/flink-streaming/flink-streaming-1.5.1-1.5.1-bin-hadoop27-scala_2.11-1531485329/lib/slf4j-log4j12-1.7.7.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> SLF4J: Found binding in
> [jar:file:/usr/hdp/2.4.2.10-1/hadoop/lib/slf4j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an
> explanation.
> SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
> 2018-07-18 12:47:03,747 INFO
> org.apache.hadoop.yarn.client.api.impl.TimelineClientImpl - Timeline service
> address: http://hmaster-1.ipbl.rgcloud.net:8188/ws/v1/timeline/
> 2018-07-18 12:47:04,222 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli -
> No path for the flink jar passed. Using the location of class
> org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
> 2018-07-18 12:47:04,222 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli -
> No path for the flink jar passed. Using the location of class
> org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
> 2018-07-18 12:47:04,248 WARN
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Neither the
> HADOOP_CONF_DIR nor the YARN_CONF_DIR environment variable is set. The Flink
> YARN Client needs one of these to be set to properly load the Hadoop
> configuration for accessing YARN.
> 2018-07-18 12:47:04,409 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Cluster specification:
> ClusterSpecification{masterMemoryMB=768, taskManagerMemoryMB=2048,
> numberTaskManagers=1, slotsPerTaskManager=1}
> 2018-07-18 12:47:04,783 WARN
> org.apache.hadoop.hdfs.shortcircuit.DomainSocketFactory - The short-circuit
> local reads feature cannot be used because libhadoop cannot be loaded.
> 2018-07-18 12:47:04,788 WARN
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - The configuration
> directory
> ('/opt/flink-streaming/flink-streaming-1.5.1-1.5.1-bin-hadoop27-scala_2.11-1531485329/conf')
> contains both LOG4J and Logback configuration files. Please delete or rename
> one of them.
> 2018-07-18 12:47:07,846 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Submitting application
> master application_1531474158783_10814
> 2018-07-18 12:47:08,073 INFO
> org.apache.hadoop.yarn.client.api.impl.YarnClientImpl - Submitted application
> application_1531474158783_10814
> 2018-07-18 12:47:08,074 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Waiting for the cluster
> to be allocated
> 2018-07-18 12:47:08,076 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Deploying cluster,
> current state ACCEPTED
> 2018-07-18 12:47:12,864 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - YARN application has
> been deployed successfully.
> {code}
> Job Manager logs:
> {code:java}
> 2018-07-18
[GitHub] GJL commented on a change in pull request #6718: [FLINK-9891] Add optional hook to shutdown cluster if a session was created in per-job mode in attached mode
GJL commented on a change in pull request #6718: [FLINK-9891] Add optional hook to shutdown cluster if a session was created in per-job mode in attached mode URL: https://github.com/apache/flink/pull/6718#discussion_r218853732 ## File path: flink-clients/src/main/java/org/apache/flink/client/cli/ProgramOptions.java ## @@ -57,6 +58,8 @@ private final boolean detachedMode; + private final boolean schutdownOnAttachedExist; Review comment: _shutdown_ This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] azagrebin commented on a change in pull request #6707: [FLINK-10157] [State TTL] Allow `null` user values in map state with TTL
azagrebin commented on a change in pull request #6707: [FLINK-10157] [State
TTL] Allow `null` user values in map state with TTL
URL: https://github.com/apache/flink/pull/6707#discussion_r218853515
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlMapState.java
##
@@ -54,7 +54,13 @@
@Override
public UV get(UK key) throws Exception {
- return getWithTtlCheckAndUpdate(() -> original.get(key), v ->
original.put(key, v), () -> original.remove(key));
+ TtlValue ttlValue = getWrapped(key);
+ return ttlValue == null ? null : ttlValue.getUserValue();
+ }
+
+ private TtlValue getWrapped(UK key) throws Exception {
+ return getWrappedWithTtlCheckAndUpdate(
+ () -> original.get(key), v -> original.put(key, v), ()
-> original.remove(key));
Review comment:
I added it for code reuse because `get` and `contains` needed the same code.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] GJL commented on a change in pull request #6718: [FLINK-9891] Add optional hook to shutdown cluster if a session was created in per-job mode in attached mode
GJL commented on a change in pull request #6718: [FLINK-9891] Add optional hook to shutdown cluster if a session was created in per-job mode in attached mode URL: https://github.com/apache/flink/pull/6718#discussion_r218853427 ## File path: docs/ops/cli.md ## @@ -257,6 +257,9 @@ Action "run" compiles and runs a program. -s,--fromSavepointPath to a savepoint to restore the job from (for example hdfs:///flink/savepoint-1537). + -sae,--schutdownOnAttachedExist If the job mode is not detached and Review comment: _shutdown_ This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-9891) Flink cluster is not shutdown in YARN mode when Flink client is stopped
[
https://issues.apache.org/jira/browse/FLINK-9891?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620750#comment-16620750
]
ASF GitHub Bot commented on FLINK-9891:
---
azagrebin commented on issue #6718: [FLINK-9891] Add optional hook to shutdown
cluster if a session was created in per-job mode in attached mode
URL: https://github.com/apache/flink/pull/6718#issuecomment-422847652
cc @tillrohrmann @GJL
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Flink cluster is not shutdown in YARN mode when Flink client is stopped
> ---
>
> Key: FLINK-9891
> URL: https://issues.apache.org/jira/browse/FLINK-9891
> Project: Flink
> Issue Type: Bug
>Affects Versions: 1.5.0, 1.5.1
>Reporter: Sergey Krasovskiy
>Assignee: Shuyi Chen
>Priority: Major
> Labels: pull-request-available
>
> We are not using session mode and detached mode. The command to run Flink job
> on YARN is:
> {code:java}
> /bin/flink run -m yarn-cluster -yn 1 -yqu flink -yjm 768 -ytm
> 2048 -j ./flink-quickstart-java-1.0-SNAPSHOT.jar -c org.test.WordCount
> {code}
> Flink CLI logs:
> {code:java}
> Setting HADOOP_CONF_DIR=/etc/hadoop/conf because no HADOOP_CONF_DIR was set.
> SLF4J: Class path contains multiple SLF4J bindings.
> SLF4J: Found binding in
> [jar:file:/opt/flink-streaming/flink-streaming-1.5.1-1.5.1-bin-hadoop27-scala_2.11-1531485329/lib/slf4j-log4j12-1.7.7.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> SLF4J: Found binding in
> [jar:file:/usr/hdp/2.4.2.10-1/hadoop/lib/slf4j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an
> explanation.
> SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
> 2018-07-18 12:47:03,747 INFO
> org.apache.hadoop.yarn.client.api.impl.TimelineClientImpl - Timeline service
> address: http://hmaster-1.ipbl.rgcloud.net:8188/ws/v1/timeline/
> 2018-07-18 12:47:04,222 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli -
> No path for the flink jar passed. Using the location of class
> org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
> 2018-07-18 12:47:04,222 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli -
> No path for the flink jar passed. Using the location of class
> org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
> 2018-07-18 12:47:04,248 WARN
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Neither the
> HADOOP_CONF_DIR nor the YARN_CONF_DIR environment variable is set. The Flink
> YARN Client needs one of these to be set to properly load the Hadoop
> configuration for accessing YARN.
> 2018-07-18 12:47:04,409 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Cluster specification:
> ClusterSpecification{masterMemoryMB=768, taskManagerMemoryMB=2048,
> numberTaskManagers=1, slotsPerTaskManager=1}
> 2018-07-18 12:47:04,783 WARN
> org.apache.hadoop.hdfs.shortcircuit.DomainSocketFactory - The short-circuit
> local reads feature cannot be used because libhadoop cannot be loaded.
> 2018-07-18 12:47:04,788 WARN
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - The configuration
> directory
> ('/opt/flink-streaming/flink-streaming-1.5.1-1.5.1-bin-hadoop27-scala_2.11-1531485329/conf')
> contains both LOG4J and Logback configuration files. Please delete or rename
> one of them.
> 2018-07-18 12:47:07,846 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Submitting application
> master application_1531474158783_10814
> 2018-07-18 12:47:08,073 INFO
> org.apache.hadoop.yarn.client.api.impl.YarnClientImpl - Submitted application
> application_1531474158783_10814
> 2018-07-18 12:47:08,074 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Waiting for the cluster
> to be allocated
> 2018-07-18 12:47:08,076 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Deploying cluster,
> current state ACCEPTED
> 2018-07-18 12:47:12,864 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - YARN application has
> been deployed successfully.
> {code}
> Job Manager logs:
> {code:java}
> 2018-07-18 12:47:09,913 INFO
> org.apache.flink.runtime.entrypoint.ClusterEntrypoint -
>
> 2018-07-18 12:47:09,915 INFO
> org.apache.flink.runtime.entrypoint.ClusterEntrypoint - Starting
> YarnSessionClusterEntrypoint (Version: 1.5.1, Rev:3488f8b, Date:10.07.2018 @
> 11:51:27 GMT)
> ...
> {code}
> Issues:
> # Flink job is running as a
[jira] [Commented] (FLINK-9891) Flink cluster is not shutdown in YARN mode when Flink client is stopped
[
https://issues.apache.org/jira/browse/FLINK-9891?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620752#comment-16620752
]
ASF GitHub Bot commented on FLINK-9891:
---
azagrebin edited a comment on issue #6718: [FLINK-9891] Add optional hook to
shutdown cluster if a session was created in per-job mode in attached mode
URL: https://github.com/apache/flink/pull/6718#issuecomment-422847652
cc @tillrohrmann @GJL @packet23
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Flink cluster is not shutdown in YARN mode when Flink client is stopped
> ---
>
> Key: FLINK-9891
> URL: https://issues.apache.org/jira/browse/FLINK-9891
> Project: Flink
> Issue Type: Bug
>Affects Versions: 1.5.0, 1.5.1
>Reporter: Sergey Krasovskiy
>Assignee: Shuyi Chen
>Priority: Major
> Labels: pull-request-available
>
> We are not using session mode and detached mode. The command to run Flink job
> on YARN is:
> {code:java}
> /bin/flink run -m yarn-cluster -yn 1 -yqu flink -yjm 768 -ytm
> 2048 -j ./flink-quickstart-java-1.0-SNAPSHOT.jar -c org.test.WordCount
> {code}
> Flink CLI logs:
> {code:java}
> Setting HADOOP_CONF_DIR=/etc/hadoop/conf because no HADOOP_CONF_DIR was set.
> SLF4J: Class path contains multiple SLF4J bindings.
> SLF4J: Found binding in
> [jar:file:/opt/flink-streaming/flink-streaming-1.5.1-1.5.1-bin-hadoop27-scala_2.11-1531485329/lib/slf4j-log4j12-1.7.7.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> SLF4J: Found binding in
> [jar:file:/usr/hdp/2.4.2.10-1/hadoop/lib/slf4j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an
> explanation.
> SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
> 2018-07-18 12:47:03,747 INFO
> org.apache.hadoop.yarn.client.api.impl.TimelineClientImpl - Timeline service
> address: http://hmaster-1.ipbl.rgcloud.net:8188/ws/v1/timeline/
> 2018-07-18 12:47:04,222 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli -
> No path for the flink jar passed. Using the location of class
> org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
> 2018-07-18 12:47:04,222 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli -
> No path for the flink jar passed. Using the location of class
> org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
> 2018-07-18 12:47:04,248 WARN
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Neither the
> HADOOP_CONF_DIR nor the YARN_CONF_DIR environment variable is set. The Flink
> YARN Client needs one of these to be set to properly load the Hadoop
> configuration for accessing YARN.
> 2018-07-18 12:47:04,409 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Cluster specification:
> ClusterSpecification{masterMemoryMB=768, taskManagerMemoryMB=2048,
> numberTaskManagers=1, slotsPerTaskManager=1}
> 2018-07-18 12:47:04,783 WARN
> org.apache.hadoop.hdfs.shortcircuit.DomainSocketFactory - The short-circuit
> local reads feature cannot be used because libhadoop cannot be loaded.
> 2018-07-18 12:47:04,788 WARN
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - The configuration
> directory
> ('/opt/flink-streaming/flink-streaming-1.5.1-1.5.1-bin-hadoop27-scala_2.11-1531485329/conf')
> contains both LOG4J and Logback configuration files. Please delete or rename
> one of them.
> 2018-07-18 12:47:07,846 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Submitting application
> master application_1531474158783_10814
> 2018-07-18 12:47:08,073 INFO
> org.apache.hadoop.yarn.client.api.impl.YarnClientImpl - Submitted application
> application_1531474158783_10814
> 2018-07-18 12:47:08,074 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Waiting for the cluster
> to be allocated
> 2018-07-18 12:47:08,076 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Deploying cluster,
> current state ACCEPTED
> 2018-07-18 12:47:12,864 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - YARN application has
> been deployed successfully.
> {code}
> Job Manager logs:
> {code:java}
> 2018-07-18 12:47:09,913 INFO
> org.apache.flink.runtime.entrypoint.ClusterEntrypoint -
>
> 2018-07-18 12:47:09,915 INFO
> org.apache.flink.runtime.entrypoint.ClusterEntrypoint - Starting
> YarnSessionClusterEntrypoint (Version: 1.5.1, Rev:3488f8b, Date:10.07.2018 @
> 11:51:27 GMT)
> ...
> {code}
> Issues:
> # Flink job is
[GitHub] azagrebin edited a comment on issue #6718: [FLINK-9891] Add optional hook to shutdown cluster if a session was created in per-job mode in attached mode
azagrebin edited a comment on issue #6718: [FLINK-9891] Add optional hook to shutdown cluster if a session was created in per-job mode in attached mode URL: https://github.com/apache/flink/pull/6718#issuecomment-422847652 cc @tillrohrmann @GJL @packet23 This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] azagrebin commented on issue #6718: [FLINK-9891] Add optional hook to shutdown cluster if a session was created in per-job mode in attached mode
azagrebin commented on issue #6718: [FLINK-9891] Add optional hook to shutdown cluster if a session was created in per-job mode in attached mode URL: https://github.com/apache/flink/pull/6718#issuecomment-422847652 cc @tillrohrmann @GJL This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-9891) Flink cluster is not shutdown in YARN mode when Flink client is stopped
[
https://issues.apache.org/jira/browse/FLINK-9891?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620749#comment-16620749
]
ASF GitHub Bot commented on FLINK-9891:
---
azagrebin opened a new pull request #6718: [FLINK-9891] Add optional hook to
shutdown cluster if a session was created in per-job mode in attached mode
URL: https://github.com/apache/flink/pull/6718
## What is the purpose of the change
This PR picks up #6540, adds a cli option to activate the cluster shutdown
hook. By default, it preserves previous behaviour where cluster is not shutdown
in case of abrupt cli exit.
## Brief change log
- add SHUTDOWN_IF_ATTACHED_OPTION in cli
## Verifying this change
Trivial option, can be checked running example job in local yarn cluster:
HADOOP_CLASSPATH=`hadoop classpath` $FLINK/bin/flink run -m yarn-cluster
$FLINK/examples/streaming/WordCount.jar
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with
`@Public(Evolving)`: (no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its
components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (yes)
- If yes, how is the feature documented? (docs)
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Flink cluster is not shutdown in YARN mode when Flink client is stopped
> ---
>
> Key: FLINK-9891
> URL: https://issues.apache.org/jira/browse/FLINK-9891
> Project: Flink
> Issue Type: Bug
>Affects Versions: 1.5.0, 1.5.1
>Reporter: Sergey Krasovskiy
>Assignee: Shuyi Chen
>Priority: Major
> Labels: pull-request-available
>
> We are not using session mode and detached mode. The command to run Flink job
> on YARN is:
> {code:java}
> /bin/flink run -m yarn-cluster -yn 1 -yqu flink -yjm 768 -ytm
> 2048 -j ./flink-quickstart-java-1.0-SNAPSHOT.jar -c org.test.WordCount
> {code}
> Flink CLI logs:
> {code:java}
> Setting HADOOP_CONF_DIR=/etc/hadoop/conf because no HADOOP_CONF_DIR was set.
> SLF4J: Class path contains multiple SLF4J bindings.
> SLF4J: Found binding in
> [jar:file:/opt/flink-streaming/flink-streaming-1.5.1-1.5.1-bin-hadoop27-scala_2.11-1531485329/lib/slf4j-log4j12-1.7.7.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> SLF4J: Found binding in
> [jar:file:/usr/hdp/2.4.2.10-1/hadoop/lib/slf4j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an
> explanation.
> SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
> 2018-07-18 12:47:03,747 INFO
> org.apache.hadoop.yarn.client.api.impl.TimelineClientImpl - Timeline service
> address: http://hmaster-1.ipbl.rgcloud.net:8188/ws/v1/timeline/
> 2018-07-18 12:47:04,222 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli -
> No path for the flink jar passed. Using the location of class
> org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
> 2018-07-18 12:47:04,222 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli -
> No path for the flink jar passed. Using the location of class
> org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
> 2018-07-18 12:47:04,248 WARN
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Neither the
> HADOOP_CONF_DIR nor the YARN_CONF_DIR environment variable is set. The Flink
> YARN Client needs one of these to be set to properly load the Hadoop
> configuration for accessing YARN.
> 2018-07-18 12:47:04,409 INFO
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Cluster specification:
> ClusterSpecification{masterMemoryMB=768, taskManagerMemoryMB=2048,
> numberTaskManagers=1, slotsPerTaskManager=1}
> 2018-07-18 12:47:04,783 WARN
> org.apache.hadoop.hdfs.shortcircuit.DomainSocketFactory - The short-circuit
> local reads feature cannot be used because libhadoop cannot be loaded.
> 2018-07-18 12:47:04,788 WARN
> org.apache.flink.yarn.AbstractYarnClusterDescriptor - The configuration
> directory
> ('/opt/flink-streaming/flink-streaming-1.5.1-1.5.1-bin-hadoop27-scala_2.11-1531485329/conf')
> contains both LOG4J and Logback configuration files. Please delete or rename
> one of them.
> 2018-07-18
[jira] [Assigned] (FLINK-10309) Cancel flink job occurs java.net.ConnectException
[
https://issues.apache.org/jira/browse/FLINK-10309?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Gary Yao reassigned FLINK-10309:
Assignee: Gary Yao
> Cancel flink job occurs java.net.ConnectException
> -
>
> Key: FLINK-10309
> URL: https://issues.apache.org/jira/browse/FLINK-10309
> Project: Flink
> Issue Type: Bug
> Components: Distributed Coordination, REST
>Affects Versions: 1.5.3, 1.6.0, 1.7.0
>Reporter: vinoyang
>Assignee: Gary Yao
>Priority: Critical
> Fix For: 1.7.0, 1.6.2, 1.5.5
>
>
> The problem occurs when using the Yarn per-job detached mode. Trying to
> cancel with savepoint fails with the following exception before being able to
> retrieve the savepoint path:
> exception stack trace :
> {code:java}
> org.apache.flink.util.FlinkException: Could not cancel job .
> at
> org.apache.flink.client.cli.CliFrontend.lambda$cancel$4(CliFrontend.java:585)
> at
> org.apache.flink.client.cli.CliFrontend.runClusterAction(CliFrontend.java:960)
> at
> org.apache.flink.client.cli.CliFrontend.cancel(CliFrontend.java:577)
> at
> org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1034)
> at java.lang.Thread.run(Thread.java:748)
> Caused by: java.util.concurrent.ExecutionException:
> org.apache.flink.runtime.concurrent.FutureUtils$RetryException: Could not
> complete the operation. Number of retries has been exhausted.
> at
> java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
> at
> java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895)
> at
> org.apache.flink.client.program.rest.RestClusterClient.cancelWithSavepoint(RestClusterClient.java:398)
> at
> org.apache.flink.client.cli.CliFrontend.lambda$cancel$4(CliFrontend.java:583)
> ... 6 more
> Caused by: org.apache.flink.runtime.concurrent.FutureUtils$RetryException:
> Could not complete the operation. Number of retries has been exhausted.
> at
> org.apache.flink.runtime.concurrent.FutureUtils.lambda$retryOperationWithDelay$5(FutureUtils.java:213)
> at
> java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:760)
> at
> java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:736)
> at
> java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474)
> at
> java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:1977)
> at
> org.apache.flink.runtime.rest.RestClient.lambda$submitRequest$1(RestClient.java:274)
> at
> org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:680)
> at
> org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListeners0(DefaultPromise.java:603)
> at
> org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:563)
> at
> org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.tryFailure(DefaultPromise.java:424)
> at
> org.apache.flink.shaded.netty4.io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.fulfillConnectPromise(AbstractNioChannel.java:268)
> at
> org.apache.flink.shaded.netty4.io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:284)
> at
> org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:528)
> at
> org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
> at
> org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
> at
> org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
> at
> org.apache.flink.shaded.netty4.io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:111)
> at
> org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137)
> ... 1 more
> Caused by: java.util.concurrent.CompletionException:
> java.net.ConnectException: Connect refuse: xxx/xxx.xxx.xxx.xxx:xxx
> at
> java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:292)
> at
> java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:308)
> at
> java.util.concurrent.CompletableFuture.uniCompose(CompletableFuture.java:943)
> at
> java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:926)
> ... 16 more
>
[GitHub] azagrebin opened a new pull request #6718: [FLINK-9891] Add optional hook to shutdown cluster if a session was created in per-job mode in attached mode
azagrebin opened a new pull request #6718: [FLINK-9891] Add optional hook to shutdown cluster if a session was created in per-job mode in attached mode URL: https://github.com/apache/flink/pull/6718 ## What is the purpose of the change This PR picks up #6540, adds a cli option to activate the cluster shutdown hook. By default, it preserves previous behaviour where cluster is not shutdown in case of abrupt cli exit. ## Brief change log - add SHUTDOWN_IF_ATTACHED_OPTION in cli ## Verifying this change Trivial option, can be checked running example job in local yarn cluster: HADOOP_CLASSPATH=`hadoop classpath` $FLINK/bin/flink run -m yarn-cluster $FLINK/examples/streaming/WordCount.jar ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): (no) - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no) - The serializers: (no) - The runtime per-record code paths (performance sensitive): (no) - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes) - The S3 file system connector: (no) ## Documentation - Does this pull request introduce a new feature? (yes) - If yes, how is the feature documented? (docs) This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-10225) Cannot access state from a empty taskmanager
[
https://issues.apache.org/jira/browse/FLINK-10225?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620743#comment-16620743
]
Pierre Zemb commented on FLINK-10225:
-
Hi! I've been reading the code a bit, and I have a question.
>From what I can see, only the TaskExecutor has enough knowledge to make a call
>to the ResourceManager. He is also the only one that is updating the
>ConcurrentHashMap used by the RPC handler. Meaning that when I'm inside the
>RPC handler, I cannot find a way to nicely trigger a method of TaskExecutor.
I like the way the interface for KvStateClientProxy is implemented, and I don't
want to change , do you have an idea on how could I implement this?
> Cannot access state from a empty taskmanager
>
>
> Key: FLINK-10225
> URL: https://issues.apache.org/jira/browse/FLINK-10225
> Project: Flink
> Issue Type: Bug
> Components: Queryable State
>Affects Versions: 1.5.3, 1.6.0
> Environment: 4tm and 1jm for now on 1.6.0
>Reporter: Pierre Zemb
>Priority: Critical
>
> Hi!
> I've started to deploy a small Flink cluster (4tm and 1jm for now on 1.6.0),
> and deployed a small job on it. Because of the current load, job is
> completely handled by a single tm. I've created a small proxy that is using
> [QueryableStateClient|https://ci.apache.org/projects/flink/flink-docs-master/api/java/org/apache/flink/queryablestate/client/QueryableStateClient.html]
> to access the current state. It is working nicely, except under certain
> circumstances. It seems to me that I can only access the state through a node
> that is holding a part of the job. Here's an example:
> * job on tm1. Pointing QueryableStateClient to tm1. State accessible
> * job still on tm1. Pointing QueryableStateClient to tm2 (for example).
> State inaccessible
> * killing tm1, job is now on tm2. State accessible
> * job still on tm2. Pointing QueryableStateClient to tm3. State inaccessible
> * adding some parallelism to spread job on tm1 and tm2. Pointing
> QueryableStateClient to either tm1 and tm2 is working
> * job still on tm1 and tm2. Pointing QueryableStateClient to tm3. State
> inaccessible
> When the state is inaccessible, I can see this (generated
> [here|https://github.com/apache/flink/blob/release-1.6/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/client/proxy/KvStateClientProxyHandler.java#L228]):
> {code:java}
> java.lang.RuntimeException: Failed request 0. Caused by:
> org.apache.flink.queryablestate.exceptions.UnknownLocationException: Could
> not retrieve location of state=repo-status of
> job=3ac3bc00b2d5bc0752917186a288d40a. Potential reasons are: i) the state is
> not ready, or ii) the job does not exist. at
> org.apache.flink.queryablestate.client.proxy.KvStateClientProxyHandler.getKvStateLookupInfo(KvStateClientProxyHandler.java:228)
> at
> org.apache.flink.queryablestate.client.proxy.KvStateClientProxyHandler.getState(KvStateClientProxyHandler.java:162)
> at
> org.apache.flink.queryablestate.client.proxy.KvStateClientProxyHandler.executeActionAsync(KvStateClientProxyHandler.java:129)
> at
> org.apache.flink.queryablestate.client.proxy.KvStateClientProxyHandler.handleRequest(KvStateClientProxyHandler.java:119)
> at
> org.apache.flink.queryablestate.client.proxy.KvStateClientProxyHandler.handleRequest(KvStateClientProxyHandler.java:63)
> at
> org.apache.flink.queryablestate.network.AbstractServerHandler$AsyncRequestTask.run(AbstractServerHandler.java:236)
> at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266) at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> at java.lang.Thread.run(Thread.java:745)
> {code}
> Went a bit through the (master branch) code. Class KvStateClientProxy is
> holding {color:#33}kvStateLocationOracle the key-value state location
> oracle for the given JobID. Here's the usage{color}{color:#33}:{color}
> * {color:#33}updateKvStateLocationOracle() in registerQueryableState()
> (TaskExecutor.java){color}
> * {color:#33}registerQueryableState() in associateWithJobManager()
> (TaskExecutor.java){color}
> * {color:#33}associateWithJobManager in establishJobManagerConnection
> (TaskExecutor.java){color}
> * {color:#33}establishJobManagerConnection in jobManagerGainedLeadership
> (TaskExecutor.java){color}
> * {color:#33}jobManagerGainedLeadership in onRegistrationSuccess
> (JobLeaderService.java){color}
> {color:#33}It seems that the KvStateLocationOracle map is updated only
> when the task manager is part of the job. {color}
> {color:#33}For now,
[jira] [Commented] (FLINK-10355) The order of the column should start from 1.
[ https://issues.apache.org/jira/browse/FLINK-10355?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620643#comment-16620643 ] lihongli commented on FLINK-10355: -- [~hequn8128] I insert a wrong row into oracle.It prints the whole row.The oracle indicates the wrong column with a * below it. > The order of the column should start from 1. > > > Key: FLINK-10355 > URL: https://issues.apache.org/jira/browse/FLINK-10355 > Project: Flink > Issue Type: Improvement > Components: Table API SQL >Affects Versions: 1.6.0 >Reporter: lihongli >Priority: Major > Labels: easyfix, pull-request-available > Attachments: B0C32FD9-47FE-4F63-921F-A9E49C0CB5CD.png > > > When I register an external Table using a CsvTableSource.It throws an > exception :"Parsing error for column 1".But I finally found that the second > column is the error column.I think that the order of the column should start > from 1. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10369) Enable YARNITCase
[
https://issues.apache.org/jira/browse/FLINK-10369?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620635#comment-16620635
]
ASF GitHub Bot commented on FLINK-10369:
TisonKun commented on issue #6717: [FLINK-10369][tests] Enable YARNITCase to
test per job mode deployment
URL: https://github.com/apache/flink/pull/6717#issuecomment-422816984
Attach the NoDataSource solution.
[patch.txt](https://github.com/apache/flink/files/2397364/patch.txt) diff to
master
[YARNITCase.java](https://github.com/apache/flink/files/2397365/YARNITCase.txt)
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Enable YARNITCase
> -
>
> Key: FLINK-10369
> URL: https://issues.apache.org/jira/browse/FLINK-10369
> Project: Flink
> Issue Type: Improvement
> Components: Tests
>Affects Versions: 1.5.3, 1.6.0, 1.7.0
>Reporter: Till Rohrmann
>Assignee: Till Rohrmann
>Priority: Minor
> Labels: pull-request-available
> Fix For: 1.7.0
>
>
> The {{YARNITCase}} is ignored because when it was added it was not possible
> to terminate the Flink cluster. This has changed now and consequently, we
> should enable this test.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[GitHub] TisonKun commented on issue #6717: [FLINK-10369][tests] Enable YARNITCase to test per job mode deployment
TisonKun commented on issue #6717: [FLINK-10369][tests] Enable YARNITCase to test per job mode deployment URL: https://github.com/apache/flink/pull/6717#issuecomment-422816984 Attach the NoDataSource solution. [patch.txt](https://github.com/apache/flink/files/2397364/patch.txt) diff to master [YARNITCase.java](https://github.com/apache/flink/files/2397365/YARNITCase.txt) This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Created] (FLINK-10371) Allow to enable SSL mutual authentication on REST endpoints by configuration
Johannes Dillmann created FLINK-10371: - Summary: Allow to enable SSL mutual authentication on REST endpoints by configuration Key: FLINK-10371 URL: https://issues.apache.org/jira/browse/FLINK-10371 Project: Flink Issue Type: Improvement Components: Client, REST, Security Affects Versions: 1.6.0, 1.7.0 Reporter: Johannes Dillmann Fix For: 1.6.1, 1.7.0 With Flink 1.6 SSL mutual authentication was introduced for internal connectivity in FLINK-9312. SSL support for external connectivity was also introduced in regard to encryption of the connection and verification of the Flink REST endpoint from the client side. But _mutual authentication between the REST endpoint and clients is not supported yet_. The [documentation suggests |https://ci.apache.org/projects/flink/flink-docs-release-1.6/ops/security-ssl.html] using a side car proxy to enable SSL mutual auth on the REST endpoint and points out the advantages of using a feature rich proxy. While this is a good rationale, there are still important use cases for support of simple mutual authentication directly in Flink: Mainly support for using standard images in a containerized environment. There are tools used to setup Flink Jobs (for example on Kubernetes clusters) and act as gateways to the Flink REST endpoint and the Flink web interface. To prevent unauthorised access to Flink the connectivity has to be secured. As the tools acts as gateway it is easy to create and pass a shared keystore and truststore used for mutual authentication to the Flink instances configurations. To enable for SSL mutual authentication on REST endpoints, I am suggesting to add a the configuration parameter `security.ssl.rest.authentication-enabled` which defaults to `false`. If it is set to `true` the `SSLUtils` factories for creating the REST server endpoint and the REST clients should set authentication to required and share `security.ssl.rest.keystore` and `security.ssl.rest.truststore` to setup SSL mutual authenticated connections. I have a working prototype which I would gladly submit as a PR to get further feedback. The changes to Flink are minimal and the default behaviour won't change. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10369) Enable YARNITCase
[
https://issues.apache.org/jira/browse/FLINK-10369?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620616#comment-16620616
]
ASF GitHub Bot commented on FLINK-10369:
TisonKun commented on a change in pull request #6717: [FLINK-10369][tests]
Enable YARNITCase to test per job mode deployment
URL: https://github.com/apache/flink/pull/6717#discussion_r218803183
##
File path: flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNITCase.java
##
@@ -79,20 +86,45 @@ public void testPerJobMode() throws Exception {
env.addSource(new InfiniteSource())
.shuffle()
- .addSink(new DiscardingSink());
+ .addSink(new DiscardingSink<>());
final JobGraph jobGraph =
env.getStreamGraph().getJobGraph();
File testingJar = YarnTestBase.findFile("..", new
TestingYarnClusterDescriptor.TestJarFinder("flink-yarn-tests"));
Review comment:
`TestingYarnClusterDescriptor` is inherited from
`LegacyYarnClusterDescriptor` and should be removed. Here we rely on
`TestingYarnClusterDescriptor.TestJarFinder`, which is a util class, and could
move it into `YarnTestUtils` to release such dependency.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Enable YARNITCase
> -
>
> Key: FLINK-10369
> URL: https://issues.apache.org/jira/browse/FLINK-10369
> Project: Flink
> Issue Type: Improvement
> Components: Tests
>Affects Versions: 1.5.3, 1.6.0, 1.7.0
>Reporter: Till Rohrmann
>Assignee: Till Rohrmann
>Priority: Minor
> Labels: pull-request-available
> Fix For: 1.7.0
>
>
> The {{YARNITCase}} is ignored because when it was added it was not possible
> to terminate the Flink cluster. This has changed now and consequently, we
> should enable this test.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-10369) Enable YARNITCase
[
https://issues.apache.org/jira/browse/FLINK-10369?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620615#comment-16620615
]
ASF GitHub Bot commented on FLINK-10369:
TisonKun commented on a change in pull request #6717: [FLINK-10369][tests]
Enable YARNITCase to test per job mode deployment
URL: https://github.com/apache/flink/pull/6717#discussion_r218805970
##
File path: flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNITCase.java
##
@@ -79,20 +86,45 @@ public void testPerJobMode() throws Exception {
env.addSource(new InfiniteSource())
.shuffle()
- .addSink(new DiscardingSink());
+ .addSink(new DiscardingSink<>());
final JobGraph jobGraph =
env.getStreamGraph().getJobGraph();
File testingJar = YarnTestBase.findFile("..", new
TestingYarnClusterDescriptor.TestJarFinder("flink-yarn-tests"));
jobGraph.addJar(new
org.apache.flink.core.fs.Path(testingJar.toURI()));
- ClusterClient clusterClient =
yarnClusterDescriptor.deployJobCluster(
- clusterSpecification,
- jobGraph,
- true);
+ ApplicationId applicationId = null;
+ ClusterClient clusterClient = null;
+
+ try {
+ clusterClient =
yarnClusterDescriptor.deployJobCluster(
+ clusterSpecification,
+ jobGraph,
+ false);
+ applicationId = clusterClient.getClusterId();
+
+ assertThat(clusterClient,
is(instanceOf(RestClusterClient.class)));
+ final RestClusterClient
restClusterClient = (RestClusterClient) clusterClient;
+
+ final CompletableFuture jobStatus =
restClusterClient.getJobStatus(jobGraph.getJobID());
+ assertThat(jobStatus.get(),
is(JobStatus.RUNNING));
Review comment:
Should we wait for `RUNNING`? or say would it be unstable when we request
jobStatus before the job switch to RUNNING?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Enable YARNITCase
> -
>
> Key: FLINK-10369
> URL: https://issues.apache.org/jira/browse/FLINK-10369
> Project: Flink
> Issue Type: Improvement
> Components: Tests
>Affects Versions: 1.5.3, 1.6.0, 1.7.0
>Reporter: Till Rohrmann
>Assignee: Till Rohrmann
>Priority: Minor
> Labels: pull-request-available
> Fix For: 1.7.0
>
>
> The {{YARNITCase}} is ignored because when it was added it was not possible
> to terminate the Flink cluster. This has changed now and consequently, we
> should enable this test.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[GitHub] TisonKun commented on a change in pull request #6717: [FLINK-10369][tests] Enable YARNITCase to test per job mode deployment
TisonKun commented on a change in pull request #6717: [FLINK-10369][tests]
Enable YARNITCase to test per job mode deployment
URL: https://github.com/apache/flink/pull/6717#discussion_r218805970
##
File path: flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNITCase.java
##
@@ -79,20 +86,45 @@ public void testPerJobMode() throws Exception {
env.addSource(new InfiniteSource())
.shuffle()
- .addSink(new DiscardingSink());
+ .addSink(new DiscardingSink<>());
final JobGraph jobGraph =
env.getStreamGraph().getJobGraph();
File testingJar = YarnTestBase.findFile("..", new
TestingYarnClusterDescriptor.TestJarFinder("flink-yarn-tests"));
jobGraph.addJar(new
org.apache.flink.core.fs.Path(testingJar.toURI()));
- ClusterClient clusterClient =
yarnClusterDescriptor.deployJobCluster(
- clusterSpecification,
- jobGraph,
- true);
+ ApplicationId applicationId = null;
+ ClusterClient clusterClient = null;
+
+ try {
+ clusterClient =
yarnClusterDescriptor.deployJobCluster(
+ clusterSpecification,
+ jobGraph,
+ false);
+ applicationId = clusterClient.getClusterId();
+
+ assertThat(clusterClient,
is(instanceOf(RestClusterClient.class)));
+ final RestClusterClient
restClusterClient = (RestClusterClient) clusterClient;
+
+ final CompletableFuture jobStatus =
restClusterClient.getJobStatus(jobGraph.getJobID());
+ assertThat(jobStatus.get(),
is(JobStatus.RUNNING));
Review comment:
Should we wait for `RUNNING`? or say would it be unstable when we request
jobStatus before the job switch to RUNNING?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] TisonKun commented on a change in pull request #6717: [FLINK-10369][tests] Enable YARNITCase to test per job mode deployment
TisonKun commented on a change in pull request #6717: [FLINK-10369][tests]
Enable YARNITCase to test per job mode deployment
URL: https://github.com/apache/flink/pull/6717#discussion_r218803183
##
File path: flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNITCase.java
##
@@ -79,20 +86,45 @@ public void testPerJobMode() throws Exception {
env.addSource(new InfiniteSource())
.shuffle()
- .addSink(new DiscardingSink());
+ .addSink(new DiscardingSink<>());
final JobGraph jobGraph =
env.getStreamGraph().getJobGraph();
File testingJar = YarnTestBase.findFile("..", new
TestingYarnClusterDescriptor.TestJarFinder("flink-yarn-tests"));
Review comment:
`TestingYarnClusterDescriptor` is inherited from
`LegacyYarnClusterDescriptor` and should be removed. Here we rely on
`TestingYarnClusterDescriptor.TestJarFinder`, which is a util class, and could
move it into `YarnTestUtils` to release such dependency.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[jira] [Updated] (FLINK-10370) DistributedCache does not work in job cluster mode.
[
https://issues.apache.org/jira/browse/FLINK-10370?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Dawid Wysakowicz updated FLINK-10370:
-
Component/s: Cluster Management
> DistributedCache does not work in job cluster mode.
> ---
>
> Key: FLINK-10370
> URL: https://issues.apache.org/jira/browse/FLINK-10370
> Project: Flink
> Issue Type: Bug
> Components: Cluster Management, Job-Submission
>Affects Versions: 1.6.0
>Reporter: Dawid Wysakowicz
>Priority: Major
>
> When using job cluster mode the client does not follow a standard submission
> path during which {{DistributedCacheEntries}} are written into
> {{Configuration}}. Therefore the files cannot be accessed in the job.
> How to reproduce:
> Simple job that uses {{DistributedCache}}:
> {code}
> public class DistributedCacheViaDfsTestProgram {
> public static void main(String[] args) throws Exception {
> final ParameterTool params = ParameterTool.fromArgs(args);
> final String inputFile =
> "hdfs://172.17.0.2:8020/home/hadoop-user/in";
> final String outputFile = "/tmp/out";
> final StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment();
> env.setParallelism(1);
> env.registerCachedFile(inputFile, "test_data", false);
> env.fromElements(1)
> .map(new TestMapFunction())
> .writeAsText(outputFile,
> FileSystem.WriteMode.OVERWRITE);
> env.execute("Distributed Cache Via Blob Test Program");
> }
> static class TestMapFunction extends RichMapFunction {
> @Override
> public String map(Integer value) throws Exception {
> final Path testFile =
> getRuntimeContext().getDistributedCache().getFile("test_data").toPath();
> return Files.readAllLines(testFile)
> .stream()
> .collect(Collectors.joining("\n"));
> }
> }
> }
> {code}
> If one runs this program e.g. in yarn job cluster mode this will produce:
> {code}
> java.lang.IllegalArgumentException: File with name 'test_data' is not
> available. Did you forget to register the file?
> at
> org.apache.flink.api.common.cache.DistributedCache.getFile(DistributedCache.java:110)
> at
> org.apache.flink.streaming.tests.DistributedCacheViaDfsTestProgram$TestMapFunction.map(DistributedCacheViaDfsTestProgram.java:59)
> at
> org.apache.flink.streaming.tests.DistributedCacheViaDfsTestProgram$TestMapFunction.map(DistributedCacheViaDfsTestProgram.java:55)
> at
> org.apache.flink.streaming.api.operators.StreamMap.processElement(StreamMap.java:41)
> at
> org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:579)
> at
> org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:554)
> at
> org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:534)
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:689)
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:667)
> at
> org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collect(StreamSourceContexts.java:104)
> at
> org.apache.flink.streaming.api.functions.source.FromElementsFunction.run(FromElementsFunction.java:164)
> at
> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:87)
> at
> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:56)
> at
> org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:99)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:300)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:711)
> at java.lang.Thread.run(Thread.java:748)
> {code}
> This job will run fine though, if it will be submitted to yarn-session
> cluster.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-10370) DistributedCache does not work in job cluster mode.
[
https://issues.apache.org/jira/browse/FLINK-10370?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620590#comment-16620590
]
Dawid Wysakowicz commented on FLINK-10370:
--
I have also some additional concerns regarding job cluster mode. Previously the
main method was executed on the client side (in most cases, besides web
submission). In 1.6.0 we allowed distributing local files via BlobServer, if
file registered in {{DistributedCache}} is a local one it would be uploaded
during submission. This obviously won't work in job cluster mode, as the
{{main}} method is executed on "server" side on which this file most probably
does not exist. This also applies to any other code in main method that assumes
it is run on the client side.
This is most notable in yarn as the job-cluster mode is now used by default for
{{-m yarn-cluster}}. This means that code that had such assumptions will work
in legacy mode (and session-cluster mode), but won't work in yarn-cluster.
> DistributedCache does not work in job cluster mode.
> ---
>
> Key: FLINK-10370
> URL: https://issues.apache.org/jira/browse/FLINK-10370
> Project: Flink
> Issue Type: Bug
> Components: Cluster Management, Job-Submission
>Affects Versions: 1.6.0
>Reporter: Dawid Wysakowicz
>Priority: Major
>
> When using job cluster mode the client does not follow a standard submission
> path during which {{DistributedCacheEntries}} are written into
> {{Configuration}}. Therefore the files cannot be accessed in the job.
> How to reproduce:
> Simple job that uses {{DistributedCache}}:
> {code}
> public class DistributedCacheViaDfsTestProgram {
> public static void main(String[] args) throws Exception {
> final ParameterTool params = ParameterTool.fromArgs(args);
> final String inputFile =
> "hdfs://172.17.0.2:8020/home/hadoop-user/in";
> final String outputFile = "/tmp/out";
> final StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment();
> env.setParallelism(1);
> env.registerCachedFile(inputFile, "test_data", false);
> env.fromElements(1)
> .map(new TestMapFunction())
> .writeAsText(outputFile,
> FileSystem.WriteMode.OVERWRITE);
> env.execute("Distributed Cache Via Blob Test Program");
> }
> static class TestMapFunction extends RichMapFunction {
> @Override
> public String map(Integer value) throws Exception {
> final Path testFile =
> getRuntimeContext().getDistributedCache().getFile("test_data").toPath();
> return Files.readAllLines(testFile)
> .stream()
> .collect(Collectors.joining("\n"));
> }
> }
> }
> {code}
> If one runs this program e.g. in yarn job cluster mode this will produce:
> {code}
> java.lang.IllegalArgumentException: File with name 'test_data' is not
> available. Did you forget to register the file?
> at
> org.apache.flink.api.common.cache.DistributedCache.getFile(DistributedCache.java:110)
> at
> org.apache.flink.streaming.tests.DistributedCacheViaDfsTestProgram$TestMapFunction.map(DistributedCacheViaDfsTestProgram.java:59)
> at
> org.apache.flink.streaming.tests.DistributedCacheViaDfsTestProgram$TestMapFunction.map(DistributedCacheViaDfsTestProgram.java:55)
> at
> org.apache.flink.streaming.api.operators.StreamMap.processElement(StreamMap.java:41)
> at
> org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:579)
> at
> org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:554)
> at
> org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:534)
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:689)
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:667)
> at
> org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collect(StreamSourceContexts.java:104)
> at
> org.apache.flink.streaming.api.functions.source.FromElementsFunction.run(FromElementsFunction.java:164)
> at
> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:87)
> at
> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:56)
> at
> org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:99)
> at
>
[jira] [Commented] (FLINK-10358) Flink kinesis connector could throw NPE during getRecords() call
[ https://issues.apache.org/jira/browse/FLINK-10358?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620586#comment-16620586 ] ASF GitHub Bot commented on FLINK-10358: tweise commented on issue #6708: [FLINK-10358][Kinesis-connector] fix NPE in case millisBehindLatest is null URL: https://github.com/apache/flink/pull/6708#issuecomment-422805054 Looks like it confirms to the spec (the field is not provided for dynamodb streams - wouldn't make sense either): https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_streams_GetRecords.html We are internally using the consumer to process the dynamodb CDC stream, plan is to add the complete support to the Kinesis connector soon. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Flink kinesis connector could throw NPE during getRecords() call > - > > Key: FLINK-10358 > URL: https://issues.apache.org/jira/browse/FLINK-10358 > Project: Flink > Issue Type: Bug > Components: Kinesis Connector >Reporter: Ying Xu >Assignee: Ying Xu >Priority: Major > Labels: pull-request-available > > When extending the flink kinesis connector to consume from a dynamodb stream, > it was found NPE could be thrown at > [here|https://github.com/apache/flink/blob/e3c98f05d3544d0165c2d97d2d00fcd295cef8c8/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/ShardConsumer.java#L376] > . > This is because the [getRecords > API|https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_streams_GetRecords.html] > in dynamodb streams does not return the millisBehindLatest field and has it > set to null. Null check is probably needed here. > See FLINK-4582 for the context of building dynamodb streams connector on top > of the Kinesis connector. > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[GitHub] tweise commented on issue #6708: [FLINK-10358][Kinesis-connector] fix NPE in case millisBehindLatest is null
tweise commented on issue #6708: [FLINK-10358][Kinesis-connector] fix NPE in case millisBehindLatest is null URL: https://github.com/apache/flink/pull/6708#issuecomment-422805054 Looks like it confirms to the spec (the field is not provided for dynamodb streams - wouldn't make sense either): https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_streams_GetRecords.html We are internally using the consumer to process the dynamodb CDC stream, plan is to add the complete support to the Kinesis connector soon. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Updated] (FLINK-10369) Enable YARNITCase
[
https://issues.apache.org/jira/browse/FLINK-10369?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-10369:
---
Labels: pull-request-available (was: )
> Enable YARNITCase
> -
>
> Key: FLINK-10369
> URL: https://issues.apache.org/jira/browse/FLINK-10369
> Project: Flink
> Issue Type: Improvement
> Components: Tests
>Affects Versions: 1.5.3, 1.6.0, 1.7.0
>Reporter: Till Rohrmann
>Assignee: Till Rohrmann
>Priority: Minor
> Labels: pull-request-available
> Fix For: 1.7.0
>
>
> The {{YARNITCase}} is ignored because when it was added it was not possible
> to terminate the Flink cluster. This has changed now and consequently, we
> should enable this test.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[GitHub] TisonKun edited a comment on issue #6716: [hotfix] [yarn-test] Clean up inactive test
TisonKun edited a comment on issue #6716: [hotfix] [yarn-test] Clean up inactive test URL: https://github.com/apache/flink/pull/6716#issuecomment-422799861 Ok, it is just `TestingYarnClusterDescriptor.TestJarFinder` which is a util class, we can move it out of `TestingYarnClusterDescriptor` If you take over the refactor of `YARNITCase`, I would do a rebase after that and remain this PR a cleanup. The rest could be cleaned up, after all. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] TisonKun commented on issue #6716: [hotfix] [yarn-test] Clean up inactive test
TisonKun commented on issue #6716: [hotfix] [yarn-test] Clean up inactive test URL: https://github.com/apache/flink/pull/6716#issuecomment-422799861 Ok, it is just `TestingYarnClusterDescriptor.TestJarFinder` which is a util class, we can move it out of `TestingYarnClusterDescriptor` This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-9646) ExecutionGraphCoLocationRestartTest.testConstraintsAfterRestart failed on Travis
[
https://issues.apache.org/jira/browse/FLINK-9646?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620561#comment-16620561
]
Till Rohrmann commented on FLINK-9646:
--
Another instance: https://api.travis-ci.org/v3/job/430511666/log.txt
> ExecutionGraphCoLocationRestartTest.testConstraintsAfterRestart failed on
> Travis
>
>
> Key: FLINK-9646
> URL: https://issues.apache.org/jira/browse/FLINK-9646
> Project: Flink
> Issue Type: Bug
> Components: Tests
>Affects Versions: 1.5.0, 1.6.0
>Reporter: Till Rohrmann
>Assignee: Till Rohrmann
>Priority: Critical
> Fix For: 1.7.0, 1.6.2, 1.5.5
>
>
> {{ExecutionGraphCoLocationRestartTest.testConstraintsAfterRestart}} fails on
> Travis.
> https://api.travis-ci.org/v3/job/395394863/log.txt
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Created] (FLINK-10370) DistributedCache does not work in job cluster mode.
Dawid Wysakowicz created FLINK-10370:
Summary: DistributedCache does not work in job cluster mode.
Key: FLINK-10370
URL: https://issues.apache.org/jira/browse/FLINK-10370
Project: Flink
Issue Type: Bug
Components: Job-Submission
Affects Versions: 1.6.0
Reporter: Dawid Wysakowicz
When using job cluster mode the client does not follow a standard submission
path during which {{DistributedCacheEntries}} are written into
{{Configuration}}. Therefore the files cannot be accessed in the job.
How to reproduce:
Simple job that uses {{DistributedCache}}:
{code}
public class DistributedCacheViaDfsTestProgram {
public static void main(String[] args) throws Exception {
final ParameterTool params = ParameterTool.fromArgs(args);
final String inputFile =
"hdfs://172.17.0.2:8020/home/hadoop-user/in";
final String outputFile = "/tmp/out";
final StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.registerCachedFile(inputFile, "test_data", false);
env.fromElements(1)
.map(new TestMapFunction())
.writeAsText(outputFile,
FileSystem.WriteMode.OVERWRITE);
env.execute("Distributed Cache Via Blob Test Program");
}
static class TestMapFunction extends RichMapFunction {
@Override
public String map(Integer value) throws Exception {
final Path testFile =
getRuntimeContext().getDistributedCache().getFile("test_data").toPath();
return Files.readAllLines(testFile)
.stream()
.collect(Collectors.joining("\n"));
}
}
}
{code}
If one runs this program e.g. in yarn job cluster mode this will produce:
{code}
java.lang.IllegalArgumentException: File with name 'test_data' is not
available. Did you forget to register the file?
at
org.apache.flink.api.common.cache.DistributedCache.getFile(DistributedCache.java:110)
at
org.apache.flink.streaming.tests.DistributedCacheViaDfsTestProgram$TestMapFunction.map(DistributedCacheViaDfsTestProgram.java:59)
at
org.apache.flink.streaming.tests.DistributedCacheViaDfsTestProgram$TestMapFunction.map(DistributedCacheViaDfsTestProgram.java:55)
at
org.apache.flink.streaming.api.operators.StreamMap.processElement(StreamMap.java:41)
at
org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:579)
at
org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:554)
at
org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:534)
at
org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:689)
at
org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:667)
at
org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collect(StreamSourceContexts.java:104)
at
org.apache.flink.streaming.api.functions.source.FromElementsFunction.run(FromElementsFunction.java:164)
at
org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:87)
at
org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:56)
at
org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:99)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:300)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:711)
at java.lang.Thread.run(Thread.java:748)
{code}
This job will run fine though, if it will be submitted to yarn-session cluster.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[GitHub] TisonKun commented on issue #6716: [hotfix] [yarn-test] Clean up inactive test
TisonKun commented on issue #6716: [hotfix] [yarn-test] Clean up inactive test URL: https://github.com/apache/flink/pull/6716#issuecomment-422799453 @tillrohrmann sure but I notice that it relies on `TestingYarnClusterDescriptor` which inherited from `LegacyYarnClusterDescriptor`. I see #6717 but it still use `TestingYarnClusterDescriptor`, is there something missed? This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-10369) Enable YARNITCase
[
https://issues.apache.org/jira/browse/FLINK-10369?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620553#comment-16620553
]
ASF GitHub Bot commented on FLINK-10369:
tillrohrmann opened a new pull request #6717: [FLINK-10369][tests] Enable
YARNITCase to test per job mode deployment
URL: https://github.com/apache/flink/pull/6717
## What is the purpose of the change
Enable YARNITCase to test per job mode deployment.
## Verifying this change
This change is a trivial rework / code cleanup without any test coverage.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with
`@Public(Evolving)`: (no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its
components), Checkpointing, Yarn/Mesos, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (no)
- If yes, how is the feature documented? (not applicable)
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Enable YARNITCase
> -
>
> Key: FLINK-10369
> URL: https://issues.apache.org/jira/browse/FLINK-10369
> Project: Flink
> Issue Type: Improvement
> Components: Tests
>Affects Versions: 1.5.3, 1.6.0, 1.7.0
>Reporter: Till Rohrmann
>Assignee: Till Rohrmann
>Priority: Minor
> Labels: pull-request-available
> Fix For: 1.7.0
>
>
> The {{YARNITCase}} is ignored because when it was added it was not possible
> to terminate the Flink cluster. This has changed now and consequently, we
> should enable this test.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[GitHub] tillrohrmann commented on issue #6716: [hotfix] [yarn-test] Clean up inactive test
tillrohrmann commented on issue #6716: [hotfix] [yarn-test] Clean up inactive test URL: https://github.com/apache/flink/pull/6716#issuecomment-422798581 Thanks for opening this PR @TisonKun. The `YarnITCase` is actually not using the legacy mode but the Flip-6 code path. Instead of removing it, I think we should activate it. I tried to see what's missing and it seems to work by adding the following changes: #6717. Therefore I would suggest to rather activate the test instead of removing it. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] tillrohrmann opened a new pull request #6717: [FLINK-10369][tests] Enable YARNITCase to test per job mode deployment
tillrohrmann opened a new pull request #6717: [FLINK-10369][tests] Enable YARNITCase to test per job mode deployment URL: https://github.com/apache/flink/pull/6717 ## What is the purpose of the change Enable YARNITCase to test per job mode deployment. ## Verifying this change This change is a trivial rework / code cleanup without any test coverage. ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): (no) - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no) - The serializers: (no) - The runtime per-record code paths (performance sensitive): (no) - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (no) - The S3 file system connector: (no) ## Documentation - Does this pull request introduce a new feature? (no) - If yes, how is the feature documented? (not applicable) This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Assigned] (FLINK-10369) Enable YARNITCase
[
https://issues.apache.org/jira/browse/FLINK-10369?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Till Rohrmann reassigned FLINK-10369:
-
Assignee: Till Rohrmann
> Enable YARNITCase
> -
>
> Key: FLINK-10369
> URL: https://issues.apache.org/jira/browse/FLINK-10369
> Project: Flink
> Issue Type: Improvement
> Components: Tests
>Affects Versions: 1.5.3, 1.6.0, 1.7.0
>Reporter: Till Rohrmann
>Assignee: Till Rohrmann
>Priority: Minor
> Fix For: 1.7.0
>
>
> The {{YARNITCase}} is ignored because when it was added it was not possible
> to terminate the Flink cluster. This has changed now and consequently, we
> should enable this test.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Created] (FLINK-10369) Enable YARNITCase
Till Rohrmann created FLINK-10369:
-
Summary: Enable YARNITCase
Key: FLINK-10369
URL: https://issues.apache.org/jira/browse/FLINK-10369
Project: Flink
Issue Type: Improvement
Components: Tests
Affects Versions: 1.6.0, 1.5.3, 1.7.0
Reporter: Till Rohrmann
Fix For: 1.7.0
The {{YARNITCase}} is ignored because when it was added it was not possible to
terminate the Flink cluster. This has changed now and consequently, we should
enable this test.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
***UNCHECKED*** [jira] [Commented] (FLINK-10050) Support 'allowedLateness' in CoGroupedStreams
[ https://issues.apache.org/jira/browse/FLINK-10050?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620532#comment-16620532 ] ASF GitHub Bot commented on FLINK-10050: EugeneYushin commented on issue #6646: [FLINK-10050] Support allowedLateness in CoGroupedStreams URL: https://github.com/apache/flink/pull/6646#issuecomment-422789390 Nice, CI checks have been passed. I think I have no access to Travis to relaunch builds (at least, I don't see any button there to perform such kind of operation). Thanks for cooperation This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Support 'allowedLateness' in CoGroupedStreams > - > > Key: FLINK-10050 > URL: https://issues.apache.org/jira/browse/FLINK-10050 > Project: Flink > Issue Type: Improvement > Components: Streaming >Affects Versions: 1.5.1, 1.6.0 >Reporter: eugen yushin >Priority: Major > Labels: pull-request-available, ready-to-commit, windows > > WindowedStream has a support of 'allowedLateness' feature, while > CoGroupedStreams are not. At the mean time, WindowedStream is an inner part > of CoGroupedStreams and all main functionality (like evictor/trigger/...) is > simply delegated to WindowedStream. > There's no chance to operate with late arriving data from previous steps in > cogroups (and joins). Consider the following flow: > a. read data from source1 -> aggregate data with allowed lateness > b. read data from source2 -> aggregate data with allowed lateness > c. cogroup/join streams a and b, and compare aggregated values > Step c doesn't accept any late data from steps a/b due to lack of > `allowedLateness` API call in CoGroupedStreams.java. > Scope: add method `WithWindow.allowedLateness` to Java API > (flink-streaming-java) and extend scala API (flink-streaming-scala). -- This message was sent by Atlassian JIRA (v7.6.3#76005)
***UNCHECKED*** [jira] [Commented] (FLINK-10355) The order of the column should start from 1.
[ https://issues.apache.org/jira/browse/FLINK-10355?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620530#comment-16620530 ] ASF GitHub Bot commented on FLINK-10355: lihongliustc commented on issue #6713: [FLINK-10355][table]the order of the columns start from 1 URL: https://github.com/apache/flink/pull/6713#issuecomment-422789369 @fhueske Thank you very much.I will continue to work hard to contribute to the community. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > The order of the column should start from 1. > > > Key: FLINK-10355 > URL: https://issues.apache.org/jira/browse/FLINK-10355 > Project: Flink > Issue Type: Improvement > Components: Table API SQL >Affects Versions: 1.6.0 >Reporter: lihongli >Priority: Major > Labels: easyfix, pull-request-available > Attachments: B0C32FD9-47FE-4F63-921F-A9E49C0CB5CD.png > > > When I register an external Table using a CsvTableSource.It throws an > exception :"Parsing error for column 1".But I finally found that the second > column is the error column.I think that the order of the column should start > from 1. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[GitHub] lihongliustc commented on issue #6713: [FLINK-10355][table]the order of the columns start from 1
lihongliustc commented on issue #6713: [FLINK-10355][table]the order of the columns start from 1 URL: https://github.com/apache/flink/pull/6713#issuecomment-422789369 @fhueske Thank you very much.I will continue to work hard to contribute to the community. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] EugeneYushin commented on issue #6646: [FLINK-10050] Support allowedLateness in CoGroupedStreams
EugeneYushin commented on issue #6646: [FLINK-10050] Support allowedLateness in CoGroupedStreams URL: https://github.com/apache/flink/pull/6646#issuecomment-422789390 Nice, CI checks have been passed. I think I have no access to Travis to relaunch builds (at least, I don't see any button there to perform such kind of operation). Thanks for cooperation This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-10355) The order of the column should start from 1.
[
https://issues.apache.org/jira/browse/FLINK-10355?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620520#comment-16620520
]
ASF GitHub Bot commented on FLINK-10355:
lihongliustc commented on a change in pull request #6713:
[FLINK-10355][table]the order of the columns start from 1
URL: https://github.com/apache/flink/pull/6713#discussion_r218782806
##
File path:
flink-java/src/main/java/org/apache/flink/api/java/io/RowCsvInputFormat.java
##
@@ -177,7 +177,7 @@ protected boolean parseRecord(Object[] holders, byte[]
bytes, int offset, int nu
// the error state EMPTY_COLUMN is
ignored
if (parser.getErrorState() !=
FieldParser.ParseErrorState.EMPTY_COLUMN) {
throw new
ParseException(String.format("Parsing error for column %1$s of row '%2$s'
originated by %3$s: %4$s.",
- field, new
String(bytes, offset, numBytes), parser.getClass().getSimpleName(),
parser.getErrorState()));
+ field+1, new
String(bytes, offset, numBytes), parser.getClass().getSimpleName(),
parser.getErrorState()));
Review comment:
OK，thank you very much.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> The order of the column should start from 1.
>
>
> Key: FLINK-10355
> URL: https://issues.apache.org/jira/browse/FLINK-10355
> Project: Flink
> Issue Type: Improvement
> Components: Table API SQL
>Affects Versions: 1.6.0
>Reporter: lihongli
>Priority: Major
> Labels: easyfix, pull-request-available
> Attachments: B0C32FD9-47FE-4F63-921F-A9E49C0CB5CD.png
>
>
> When I register an external Table using a CsvTableSource.It throws an
> exception :"Parsing error for column 1".But I finally found that the second
> column is the error column.I think that the order of the column should start
> from 1.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[GitHub] lihongliustc commented on a change in pull request #6713: [FLINK-10355][table]the order of the columns start from 1
lihongliustc commented on a change in pull request #6713:
[FLINK-10355][table]the order of the columns start from 1
URL: https://github.com/apache/flink/pull/6713#discussion_r218782806
##
File path:
flink-java/src/main/java/org/apache/flink/api/java/io/RowCsvInputFormat.java
##
@@ -177,7 +177,7 @@ protected boolean parseRecord(Object[] holders, byte[]
bytes, int offset, int nu
// the error state EMPTY_COLUMN is
ignored
if (parser.getErrorState() !=
FieldParser.ParseErrorState.EMPTY_COLUMN) {
throw new
ParseException(String.format("Parsing error for column %1$s of row '%2$s'
originated by %3$s: %4$s.",
- field, new
String(bytes, offset, numBytes), parser.getClass().getSimpleName(),
parser.getErrorState()));
+ field+1, new
String(bytes, offset, numBytes), parser.getClass().getSimpleName(),
parser.getErrorState()));
Review comment:
OK，thank you very much.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[jira] [Commented] (FLINK-10320) Introduce JobMaster schedule micro-benchmark
[
https://issues.apache.org/jira/browse/FLINK-10320?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620511#comment-16620511
]
陈梓立 commented on FLINK-10320:
-
We have some cases when our customers set parallelism to over 10,000 because of
the amount of data to process. In such case, {{JobMaster}} would be even
unavailable because busy to handle rpc requests or gc. This is the original
motivation to introduce a benchmark aimed at prevent regression on schedule
module.
> Introduce JobMaster schedule micro-benchmark
>
>
> Key: FLINK-10320
> URL: https://issues.apache.org/jira/browse/FLINK-10320
> Project: Flink
> Issue Type: Improvement
> Components: Tests
>Reporter: 陈梓立
>Assignee: 陈梓立
>Priority: Major
>
> Based on {{org.apache.flink.streaming.runtime.io.benchmark}} stuff and the
> repo [flink-benchmark|https://github.com/dataArtisans/flink-benchmarks], I
> proposal to introduce another micro-benchmark which focuses on {{JobMaster}}
> schedule performance
> h3. Target
> Benchmark how long from {{JobMaster}} startup(receive the {{JobGraph}} and
> init) to all tasks RUNNING. Technically we use bounded stream and TM finishes
> tasks as soon as they arrived. So the real interval we measure is to all
> tasks FINISHED.
> h3. Case
> 1. JobGraph that cover EAGER + PIPELINED edges
> 2. JobGraph that cover LAZY_FROM_SOURCES + PIPELINED edges
> 3. JobGraph that cover LAZY_FROM_SOURCES + BLOCKING edges
> ps: maybe benchmark if the source is get from {{InputSplit}}?
> h3. Implement
> Based on the flink-benchmark repo, we finally run benchmark using jmh. So the
> whole test suit is separated into two repos. The testing environment could be
> located in the main repo, maybe under
> flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/benchmark.
> To measure the performance of {{JobMaster}} scheduling, we need to simulate
> an environment that:
> 1. has a real {{JobMaster}}
> 2. has a mock/testing {{ResourceManager}} that having infinite resource and
> react immediately.
> 3. has a(many?) mock/testing {{TaskExecutor}} that deploy and finish tasks
> immediately.
> [~trohrm...@apache.org] [~GJL] [~pnowojski] could you please review this
> proposal to help clarify the goal and concrete details? Thanks in advance.
> Any suggestions are welcome.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9697) Provide connector for Kafka 2.0.0
[ https://issues.apache.org/jira/browse/FLINK-9697?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620506#comment-16620506 ] ASF GitHub Bot commented on FLINK-9697: --- yanghua commented on issue #6703: [FLINK-9697] Provide connector for Kafka 2.0.0 URL: https://github.com/apache/flink/pull/6703#issuecomment-422778285 @pnowojski If we don't consider the failed test case, then this PR applies to both Kafka client 1.0 and kafka client 2.0. Basically it also applies to kafka 0.11 (except for a reflection field of the producer transaction and the associated test code). So, I want to hear your opinion. Should we build a connector that fits the three versions of kafka client next? This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Provide connector for Kafka 2.0.0 > - > > Key: FLINK-9697 > URL: https://issues.apache.org/jira/browse/FLINK-9697 > Project: Flink > Issue Type: Improvement >Reporter: Ted Yu >Assignee: vinoyang >Priority: Major > Labels: pull-request-available > > Kafka 2.0.0 would be released soon. > Here is vote thread: > [http://search-hadoop.com/m/Kafka/uyzND1vxnEd23QLxb?subj=+VOTE+2+0+0+RC1] > We should provide connector for Kafka 2.0.0 once it is released. > Upgrade to 2.0 documentation : > http://kafka.apache.org/20/documentation.html#upgrade_2_0_0 -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[GitHub] yanghua commented on issue #6703: [FLINK-9697] Provide connector for Kafka 2.0.0
yanghua commented on issue #6703: [FLINK-9697] Provide connector for Kafka 2.0.0 URL: https://github.com/apache/flink/pull/6703#issuecomment-422778285 @pnowojski If we don't consider the failed test case, then this PR applies to both Kafka client 1.0 and kafka client 2.0. Basically it also applies to kafka 0.11 (except for a reflection field of the producer transaction and the associated test code). So, I want to hear your opinion. Should we build a connector that fits the three versions of kafka client next? This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-8424) [Cassandra Connector] Update Cassandra version to last one
[ https://issues.apache.org/jira/browse/FLINK-8424?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620503#comment-16620503 ] ASF GitHub Bot commented on FLINK-8424: --- yanghua commented on a change in pull request #6715: FLINK-8424: update cassandra and driver version to latest URL: https://github.com/apache/flink/pull/6715#discussion_r218775890 ## File path: flink-connectors/flink-connector-cassandra/pom.xml ## @@ -37,8 +37,8 @@ under the License. - 2.2.5 - 3.0.0 + 3.11.3 Review comment: when you upgrade the dependency's version, please also provide it's dependency tree about the two version. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > [Cassandra Connector] Update Cassandra version to last one > -- > > Key: FLINK-8424 > URL: https://issues.apache.org/jira/browse/FLINK-8424 > Project: Flink > Issue Type: Improvement > Components: Cassandra Connector >Reporter: Joao Boto >Priority: Major > Labels: pull-request-available > > Cassandra connector are using a version outdated > This is to upgrade the cassandra version to something new > https://git1-us-west.apache.org/repos/asf?p=cassandra.git;a=blob_plain;f=CHANGES.txt;hb=refs/tags/cassandra-3.11.1 -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[GitHub] yanghua commented on a change in pull request #6715: FLINK-8424: update cassandra and driver version to latest
yanghua commented on a change in pull request #6715: FLINK-8424: update cassandra and driver version to latest URL: https://github.com/apache/flink/pull/6715#discussion_r218775890 ## File path: flink-connectors/flink-connector-cassandra/pom.xml ## @@ -37,8 +37,8 @@ under the License. - 2.2.5 - 3.0.0 + 3.11.3 Review comment: when you upgrade the dependency's version, please also provide it's dependency tree about the two version. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
***UNCHECKED*** [jira] [Closed] (FLINK-10324) Replace ZooKeeperStateHandleStore#getAllSortedByNameAndLock by getAllAndLock
[
https://issues.apache.org/jira/browse/FLINK-10324?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Till Rohrmann closed FLINK-10324.
-
Resolution: Fixed
Fixed via e7ac3ba7dfcb90c21025def2bf4112b108d21afd
> Replace ZooKeeperStateHandleStore#getAllSortedByNameAndLock by getAllAndLock
>
>
> Key: FLINK-10324
> URL: https://issues.apache.org/jira/browse/FLINK-10324
> Project: Flink
> Issue Type: Sub-task
> Components: Distributed Coordination
>Affects Versions: 1.7.0
>Reporter: Till Rohrmann
>Assignee: Till Rohrmann
>Priority: Minor
> Labels: pull-request-available
> Fix For: 1.7.0
>
>
> In order to reduce code duplication I propose to replace
> {{ZooKeeperStateHandleStore#getAllSortedByNameAndLock}} by
> {{ZooKeeperStateHandleStore#getAllAndLock}} and sort afterwards.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[GitHub] TisonKun opened a new pull request #6716: [hotfix] [yarn-test] Clean up inactive test
TisonKun opened a new pull request #6716: [hotfix] [yarn-test] Clean up inactive test URL: https://github.com/apache/flink/pull/6716 ## What is the purpose of the change 1. YARNITCase is @Ignore 2. YARNHighAvailabilityITCase only run on legacy mode, which is inactive during Travis CI 3. After these two inactive cases removed, those testing class could be removed, too. They are inherited from legacy components and would not be supported any more. To sum up, this PR is to remove unmaintained test. cc @GJL @zentol @azagrebin This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-10324) Replace ZooKeeperStateHandleStore#getAllSortedByNameAndLock by getAllAndLock
[
https://issues.apache.org/jira/browse/FLINK-10324?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620497#comment-16620497
]
ASF GitHub Bot commented on FLINK-10324:
tillrohrmann closed pull request #6681: [FLINK-10324] Replace
ZooKeeperStateHandleStore#getAllSortedByNameAndLock by getAllAndLock
URL: https://github.com/apache/flink/pull/6681
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java
index e443fc21552..51f40085c3f 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java
@@ -35,6 +35,8 @@
import java.io.IOException;
import java.util.ArrayDeque;
import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
import java.util.ConcurrentModificationException;
import java.util.List;
import java.util.concurrent.Executor;
@@ -69,6 +71,8 @@
private static final Logger LOG =
LoggerFactory.getLogger(ZooKeeperCompletedCheckpointStore.class);
+ private static final
Comparator, String>>
STRING_COMPARATOR = Comparator.comparing(o -> o.f1);
+
/** Curator ZooKeeper client. */
private final CuratorFramework client;
@@ -153,7 +157,7 @@ public void recover() throws Exception {
List,
String>> initialCheckpoints;
while (true) {
try {
- initialCheckpoints =
checkpointsInZooKeeper.getAllSortedByNameAndLock();
+ initialCheckpoints =
checkpointsInZooKeeper.getAllAndLock();
break;
}
catch (ConcurrentModificationException e) {
@@ -161,6 +165,8 @@ public void recover() throws Exception {
}
}
+ Collections.sort(initialCheckpoints, STRING_COMPARATOR);
+
int numberOfInitialCheckpoints = initialCheckpoints.size();
LOG.info("Found {} checkpoints in ZooKeeper.",
numberOfInitialCheckpoints);
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperStateHandleStore.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperStateHandleStore.java
index b9cd0c1b720..2cb1ccc5071 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperStateHandleStore.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperStateHandleStore.java
@@ -24,7 +24,6 @@
import org.apache.flink.util.InstantiationUtil;
import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.utils.ZKPaths;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.data.Stat;
@@ -320,68 +319,6 @@ public int exists(String pathInZooKeeper) throws Exception
{
return stateHandles;
}
-
- /**
-* Gets all available state handles from ZooKeeper sorted by name
(ascending) and locks the
-* respective state nodes. The result tuples contain the retrieved
state and the path to the
-* node in ZooKeeper.
-*
-* If there is a concurrent modification, the operation is retried
until it succeeds.
-*
-* @return All state handles in ZooKeeper.
-* @throws Exception If a ZooKeeper or state handle operation fails
-*/
- @SuppressWarnings("unchecked")
- public List, String>>
getAllSortedByNameAndLock() throws Exception {
- final List, String>>
stateHandles = new ArrayList<>();
-
- boolean success = false;
-
- retry:
- while (!success) {
- stateHandles.clear();
-
- Stat stat = client.checkExists().forPath("/");
- if (stat == null) {
- break; // Node does not exist, done.
- } else {
- // Initial cVersion (number of changes to the
children of this node)
- int initialCVersion = stat.getCversion();
-
- List children =
ZKPaths.getSortedChildren(
-
client.getZookeeperClient().getZooKeeper(),
-
[GitHub] tillrohrmann closed pull request #6681: [FLINK-10324] Replace ZooKeeperStateHandleStore#getAllSortedByNameAndLock by getAllAndLock
tillrohrmann closed pull request #6681: [FLINK-10324] Replace
ZooKeeperStateHandleStore#getAllSortedByNameAndLock by getAllAndLock
URL: https://github.com/apache/flink/pull/6681
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java
index e443fc21552..51f40085c3f 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java
@@ -35,6 +35,8 @@
import java.io.IOException;
import java.util.ArrayDeque;
import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
import java.util.ConcurrentModificationException;
import java.util.List;
import java.util.concurrent.Executor;
@@ -69,6 +71,8 @@
private static final Logger LOG =
LoggerFactory.getLogger(ZooKeeperCompletedCheckpointStore.class);
+ private static final
Comparator, String>>
STRING_COMPARATOR = Comparator.comparing(o -> o.f1);
+
/** Curator ZooKeeper client. */
private final CuratorFramework client;
@@ -153,7 +157,7 @@ public void recover() throws Exception {
List,
String>> initialCheckpoints;
while (true) {
try {
- initialCheckpoints =
checkpointsInZooKeeper.getAllSortedByNameAndLock();
+ initialCheckpoints =
checkpointsInZooKeeper.getAllAndLock();
break;
}
catch (ConcurrentModificationException e) {
@@ -161,6 +165,8 @@ public void recover() throws Exception {
}
}
+ Collections.sort(initialCheckpoints, STRING_COMPARATOR);
+
int numberOfInitialCheckpoints = initialCheckpoints.size();
LOG.info("Found {} checkpoints in ZooKeeper.",
numberOfInitialCheckpoints);
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperStateHandleStore.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperStateHandleStore.java
index b9cd0c1b720..2cb1ccc5071 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperStateHandleStore.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperStateHandleStore.java
@@ -24,7 +24,6 @@
import org.apache.flink.util.InstantiationUtil;
import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.utils.ZKPaths;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.data.Stat;
@@ -320,68 +319,6 @@ public int exists(String pathInZooKeeper) throws Exception
{
return stateHandles;
}
-
- /**
-* Gets all available state handles from ZooKeeper sorted by name
(ascending) and locks the
-* respective state nodes. The result tuples contain the retrieved
state and the path to the
-* node in ZooKeeper.
-*
-* If there is a concurrent modification, the operation is retried
until it succeeds.
-*
-* @return All state handles in ZooKeeper.
-* @throws Exception If a ZooKeeper or state handle operation fails
-*/
- @SuppressWarnings("unchecked")
- public List, String>>
getAllSortedByNameAndLock() throws Exception {
- final List, String>>
stateHandles = new ArrayList<>();
-
- boolean success = false;
-
- retry:
- while (!success) {
- stateHandles.clear();
-
- Stat stat = client.checkExists().forPath("/");
- if (stat == null) {
- break; // Node does not exist, done.
- } else {
- // Initial cVersion (number of changes to the
children of this node)
- int initialCVersion = stat.getCversion();
-
- List children =
ZKPaths.getSortedChildren(
-
client.getZookeeperClient().getZooKeeper(),
-
ZKPaths.fixForNamespace(client.getNamespace(), "/"));
-
- for (String path : children) {
- path = "/" + path;
-
- try {
-
[jira] [Created] (FLINK-10368) 'Kerberized YARN on Docker test' instable
Till Rohrmann created FLINK-10368:
-
Summary: 'Kerberized YARN on Docker test' instable
Key: FLINK-10368
URL: https://issues.apache.org/jira/browse/FLINK-10368
Project: Flink
Issue Type: Bug
Components: Tests
Affects Versions: 1.6.0, 1.5.3, 1.7.0
Reporter: Till Rohrmann
Fix For: 1.7.0
Running Kerberized YARN on Docker test end-to-end test failed on an AWS
instance. The problem seems to be that the NameNode went into safe-mode due to
limited resources.
{code}
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in
[jar:file:/home/hadoop-user/flink-1.6.1/lib/slf4j-log4j12-1.7.7.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in
[jar:file:/usr/local/hadoop-2.8.4/share/hadoop/common/lib/slf4j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
2018-09-19 09:04:39,201 INFO org.apache.hadoop.security.UserGroupInformation
- Login successful for user hadoop-user using keytab file
/home/hadoop-user/hadoop-user.keytab
2018-09-19 09:04:39,453 INFO org.apache.hadoop.yarn.client.RMProxy
- Connecting to ResourceManager at
master.docker-hadoop-cluster-network/172.22.0.3:8032
2018-09-19 09:04:39,640 INFO org.apache.hadoop.yarn.client.AHSProxy
- Connecting to Application History server at
master.docker-hadoop-cluster-network/172.22.0.3:10200
2018-09-19 09:04:39,656 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli
- No path for the flink jar passed. Using the location of class
org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
2018-09-19 09:04:39,656 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli
- No path for the flink jar passed. Using the location of class
org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
2018-09-19 09:04:39,901 INFO
org.apache.flink.yarn.AbstractYarnClusterDescriptor - Cluster
specification: ClusterSpecification{masterMemoryMB=2000,
taskManagerMemoryMB=2000, numberTaskManagers=3, slotsPerTaskManager=1}
2018-09-19 09:04:40,286 WARN
org.apache.flink.yarn.AbstractYarnClusterDescriptor - The
configuration directory ('/home/hadoop-user/flink-1.6.1/conf') contains both
LOG4J and Logback configuration files. Please delete or rename one of them.
The program finished with the following exception:
org.apache.flink.client.deployment.ClusterDeploymentException: Couldn't deploy
Yarn session cluster
at
org.apache.flink.yarn.AbstractYarnClusterDescriptor.deploySessionCluster(AbstractYarnClusterDescriptor.java:420)
at
org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:259)
at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:215)
at
org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1044)
at
org.apache.flink.client.cli.CliFrontend.lambda$main$11(CliFrontend.java:1120)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:422)
at
org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1836)
at
org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41)
at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1120)
Caused by: org.apache.hadoop.hdfs.server.namenode.SafeModeException: Cannot
create
file/user/hadoop-user/.flink/application_1537266361291_0099/lib/slf4j-log4j12-1.7.7.jar.
Name node is in safe mode.
Resources are low on NN. Please add or free up more resources then turn off
safe mode manually. NOTE: If you turn off safe mode before adding resources,
the NN will immediately return to safe mode. Use "hdfs dfsadmin -safemode
leave" to turn safe mode off.
NamenodeHostName:master.docker-hadoop-cluster-network
at
org.apache.hadoop.hdfs.server.namenode.FSNamesystem.newSafemodeException(FSNamesystem.java:1407)
at
org.apache.hadoop.hdfs.server.namenode.FSNamesystem.checkNameNodeSafeMode(FSNamesystem.java:1395)
at
org.apache.hadoop.hdfs.server.namenode.FSNamesystem.startFileInt(FSNamesystem.java:2278)
at
org.apache.hadoop.hdfs.server.namenode.FSNamesystem.startFile(FSNamesystem.java:2223)
at
org.apache.hadoop.hdfs.server.namenode.NameNodeRpcServer.create(NameNodeRpcServer.java:728)
at
org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolServerSideTranslatorPB.create(ClientNamenodeProtocolServerSideTranslatorPB.java:413)
at
[GitHub] GJL commented on issue #6667: [hotfix][docs] Fix error typo in CheckpointedFunction
GJL commented on issue #6667: [hotfix][docs] Fix error typo in CheckpointedFunction URL: https://github.com/apache/flink/pull/6667#issuecomment-422770556 @lamber-ken Do you want to address the comments? This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-10332) Move data available notification in PipelinedSubpartition out of the synchronized block
[
https://issues.apache.org/jira/browse/FLINK-10332?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620483#comment-16620483
]
ASF GitHub Bot commented on FLINK-10332:
NicoK closed pull request #6693: [FLINK-10332][network] move data notification
out of the synchronized block
URL: https://github.com/apache/flink/pull/6693
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java
index 90daf75fcc7..cc0b2220fd2 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java
@@ -89,9 +89,7 @@
public void addInputChannel(RemoteInputChannel listener) throws
IOException {
checkError();
- if (!inputChannels.containsKey(listener.getInputChannelId())) {
- inputChannels.put(listener.getInputChannelId(),
listener);
- }
+ inputChannels.putIfAbsent(listener.getInputChannelId(),
listener);
}
@Override
@@ -112,12 +110,7 @@ public void cancelRequestFor(InputChannelID
inputChannelId) {
@Override
public void notifyCreditAvailable(final RemoteInputChannel
inputChannel) {
- ctx.executor().execute(new Runnable() {
- @Override
- public void run() {
-
ctx.pipeline().fireUserEventTriggered(inputChannel);
- }
- });
+ ctx.executor().execute(() ->
ctx.pipeline().fireUserEventTriggered(inputChannel));
}
//
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
index 796e86f51b3..c5ba7a4b7f1 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
@@ -85,9 +85,7 @@
public void addInputChannel(RemoteInputChannel listener) throws
IOException {
checkError();
- if (!inputChannels.containsKey(listener.getInputChannelId())) {
- inputChannels.put(listener.getInputChannelId(),
listener);
- }
+ inputChannels.putIfAbsent(listener.getInputChannelId(),
listener);
}
@Override
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
index 8c05b8208f9..c3d3d1bcc10 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
@@ -89,12 +89,7 @@ void notifyReaderNonEmpty(final NetworkSequenceViewReader
reader) {
// TODO This could potentially have a bad performance impact as
in the
// worst case (network consumes faster than the producer) each
buffer
// will trigger a separate event loop task being scheduled.
- ctx.executor().execute(new Runnable() {
- @Override
- public void run() {
- ctx.pipeline().fireUserEventTriggered(reader);
- }
- });
+ ctx.executor().execute(() ->
ctx.pipeline().fireUserEventTriggered(reader));
}
/**
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
index d2d7fdb324b..fe27d97adaa 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
@@ -88,6 +88,7 @@ public void finish() throws IOException {
private boolean add(BufferConsumer bufferConsumer, boolean
[GitHub] NicoK closed pull request #6693: [FLINK-10332][network] move data notification out of the synchronized block
NicoK closed pull request #6693: [FLINK-10332][network] move data notification
out of the synchronized block
URL: https://github.com/apache/flink/pull/6693
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java
index 90daf75fcc7..cc0b2220fd2 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java
@@ -89,9 +89,7 @@
public void addInputChannel(RemoteInputChannel listener) throws
IOException {
checkError();
- if (!inputChannels.containsKey(listener.getInputChannelId())) {
- inputChannels.put(listener.getInputChannelId(),
listener);
- }
+ inputChannels.putIfAbsent(listener.getInputChannelId(),
listener);
}
@Override
@@ -112,12 +110,7 @@ public void cancelRequestFor(InputChannelID
inputChannelId) {
@Override
public void notifyCreditAvailable(final RemoteInputChannel
inputChannel) {
- ctx.executor().execute(new Runnable() {
- @Override
- public void run() {
-
ctx.pipeline().fireUserEventTriggered(inputChannel);
- }
- });
+ ctx.executor().execute(() ->
ctx.pipeline().fireUserEventTriggered(inputChannel));
}
//
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
index 796e86f51b3..c5ba7a4b7f1 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
@@ -85,9 +85,7 @@
public void addInputChannel(RemoteInputChannel listener) throws
IOException {
checkError();
- if (!inputChannels.containsKey(listener.getInputChannelId())) {
- inputChannels.put(listener.getInputChannelId(),
listener);
- }
+ inputChannels.putIfAbsent(listener.getInputChannelId(),
listener);
}
@Override
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
index 8c05b8208f9..c3d3d1bcc10 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
@@ -89,12 +89,7 @@ void notifyReaderNonEmpty(final NetworkSequenceViewReader
reader) {
// TODO This could potentially have a bad performance impact as
in the
// worst case (network consumes faster than the producer) each
buffer
// will trigger a separate event loop task being scheduled.
- ctx.executor().execute(new Runnable() {
- @Override
- public void run() {
- ctx.pipeline().fireUserEventTriggered(reader);
- }
- });
+ ctx.executor().execute(() ->
ctx.pipeline().fireUserEventTriggered(reader));
}
/**
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
index d2d7fdb324b..fe27d97adaa 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
@@ -88,6 +88,7 @@ public void finish() throws IOException {
private boolean add(BufferConsumer bufferConsumer, boolean finish) {
checkNotNull(bufferConsumer);
+ final boolean notifyDataAvailable;
synchronized (buffers) {
if (isFinished || isReleased) {
[jira] [Commented] (FLINK-8799) Make AbstractYarnClusterDescriptor immutable
[
https://issues.apache.org/jira/browse/FLINK-8799?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620472#comment-16620472
]
ASF GitHub Bot commented on FLINK-8799:
---
yanghua closed pull request #5617: [FLINK-8799][YARN] Make
AbstractYarnClusterDescriptor immutable
URL: https://github.com/apache/flink/pull/5617
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/TestingYarnClusterDescriptor.java
b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/TestingYarnClusterDescriptor.java
index 4d2aaa02efe..ff74e4d358e 100644
---
a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/TestingYarnClusterDescriptor.java
+++
b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/TestingYarnClusterDescriptor.java
@@ -21,15 +21,12 @@
import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.jobgraph.JobGraph;
-import org.apache.flink.util.Preconditions;
import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import java.io.File;
import java.io.FilenameFilter;
-import java.util.ArrayList;
-import java.util.List;
/**
* Yarn client which starts a {@link TestingApplicationMaster}. Additionally
the client adds the
@@ -50,25 +47,6 @@ public TestingYarnClusterDescriptor(
configurationDirectory,
yarnClient,
sharedYarnClient);
- List filesToShip = new ArrayList<>();
-
- File testingJar = YarnTestBase.findFile("..", new
TestJarFinder("flink-yarn-tests"));
- Preconditions.checkNotNull(testingJar, "Could not find the
flink-yarn-tests tests jar. " +
- "Make sure to package the flink-yarn-tests module.");
-
- File testingRuntimeJar = YarnTestBase.findFile("..", new
TestJarFinder("flink-runtime"));
- Preconditions.checkNotNull(testingRuntimeJar, "Could not find
the flink-runtime tests " +
- "jar. Make sure to package the flink-runtime module.");
-
- File testingYarnJar = YarnTestBase.findFile("..", new
TestJarFinder("flink-yarn"));
- Preconditions.checkNotNull(testingRuntimeJar, "Could not find
the flink-yarn tests " +
- "jar. Make sure to package the flink-yarn module.");
-
- filesToShip.add(testingJar);
- filesToShip.add(testingRuntimeJar);
- filesToShip.add(testingYarnJar);
-
- addShipFiles(filesToShip);
}
@Override
diff --git
a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
index f9c03f93784..18e8e6b58a5 100644
---
a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
+++
b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
@@ -34,12 +34,13 @@
import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages;
import org.apache.flink.runtime.util.LeaderRetrievalUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.flink.yarn.configuration.YarnConfigOptions;
import akka.actor.ActorSystem;
import akka.actor.PoisonPill;
import akka.testkit.JavaTestKit;
import org.apache.curator.test.TestingServer;
-import org.apache.hadoop.fs.Path;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.junit.AfterClass;
@@ -50,7 +51,6 @@
import org.junit.rules.TemporaryFolder;
import java.io.File;
-import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import scala.concurrent.duration.FiniteDuration;
@@ -110,29 +110,50 @@ public void testMultipleAMKill() throws Exception {
final int numberKillingAttempts = numberApplicationAttempts - 1;
String confDirPath =
System.getenv(ConfigConstants.ENV_FLINK_CONF_DIR);
final Configuration configuration =
GlobalConfiguration.loadConfiguration();
- TestingYarnClusterDescriptor flinkYarnClient = new
TestingYarnClusterDescriptor(
- configuration,
- getYarnConfiguration(),
- confDirPath,
- getYarnClient(),
- true);
- Assert.assertNotNull("unable to get yarn client",
flinkYarnClient);
- flinkYarnClient.setLocalJarPath(new
[jira] [Commented] (FLINK-8799) Make AbstractYarnClusterDescriptor immutable
[
https://issues.apache.org/jira/browse/FLINK-8799?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620471#comment-16620471
]
ASF GitHub Bot commented on FLINK-8799:
---
yanghua commented on issue #5617: [FLINK-8799][YARN] Make
AbstractYarnClusterDescriptor immutable
URL: https://github.com/apache/flink/pull/5617#issuecomment-422764603
@GJL OK, closing this PR.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Make AbstractYarnClusterDescriptor immutable
>
>
> Key: FLINK-8799
> URL: https://issues.apache.org/jira/browse/FLINK-8799
> Project: Flink
> Issue Type: Improvement
> Components: YARN
>Affects Versions: 1.5.0
>Reporter: Gary Yao
>Priority: Major
> Labels: pull-request-available
> Fix For: 1.7.0
>
>
> {{AbstractYarnClusterDescriptor}} should be made immutable. Currently, its
> internal configuration is modified from different places which makes it
> difficult to reason about the code. For example, it should not be possible to
> modify the {{zookeeperNamespace}} using a setter method. A user of this class
> should be forced to provide all information prior to creating the instance,
> e.g., by passing a {{org.apache.flink.configuration.Configuration}} object.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[GitHub] yanghua commented on issue #5617: [FLINK-8799][YARN] Make AbstractYarnClusterDescriptor immutable
yanghua commented on issue #5617: [FLINK-8799][YARN] Make AbstractYarnClusterDescriptor immutable URL: https://github.com/apache/flink/pull/5617#issuecomment-422764603 @GJL OK, closing this PR. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] yanghua closed pull request #5617: [FLINK-8799][YARN] Make AbstractYarnClusterDescriptor immutable
yanghua closed pull request #5617: [FLINK-8799][YARN] Make
AbstractYarnClusterDescriptor immutable
URL: https://github.com/apache/flink/pull/5617
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/TestingYarnClusterDescriptor.java
b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/TestingYarnClusterDescriptor.java
index 4d2aaa02efe..ff74e4d358e 100644
---
a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/TestingYarnClusterDescriptor.java
+++
b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/TestingYarnClusterDescriptor.java
@@ -21,15 +21,12 @@
import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.jobgraph.JobGraph;
-import org.apache.flink.util.Preconditions;
import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import java.io.File;
import java.io.FilenameFilter;
-import java.util.ArrayList;
-import java.util.List;
/**
* Yarn client which starts a {@link TestingApplicationMaster}. Additionally
the client adds the
@@ -50,25 +47,6 @@ public TestingYarnClusterDescriptor(
configurationDirectory,
yarnClient,
sharedYarnClient);
- List filesToShip = new ArrayList<>();
-
- File testingJar = YarnTestBase.findFile("..", new
TestJarFinder("flink-yarn-tests"));
- Preconditions.checkNotNull(testingJar, "Could not find the
flink-yarn-tests tests jar. " +
- "Make sure to package the flink-yarn-tests module.");
-
- File testingRuntimeJar = YarnTestBase.findFile("..", new
TestJarFinder("flink-runtime"));
- Preconditions.checkNotNull(testingRuntimeJar, "Could not find
the flink-runtime tests " +
- "jar. Make sure to package the flink-runtime module.");
-
- File testingYarnJar = YarnTestBase.findFile("..", new
TestJarFinder("flink-yarn"));
- Preconditions.checkNotNull(testingRuntimeJar, "Could not find
the flink-yarn tests " +
- "jar. Make sure to package the flink-yarn module.");
-
- filesToShip.add(testingJar);
- filesToShip.add(testingRuntimeJar);
- filesToShip.add(testingYarnJar);
-
- addShipFiles(filesToShip);
}
@Override
diff --git
a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
index f9c03f93784..18e8e6b58a5 100644
---
a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
+++
b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
@@ -34,12 +34,13 @@
import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages;
import org.apache.flink.runtime.util.LeaderRetrievalUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.flink.yarn.configuration.YarnConfigOptions;
import akka.actor.ActorSystem;
import akka.actor.PoisonPill;
import akka.testkit.JavaTestKit;
import org.apache.curator.test.TestingServer;
-import org.apache.hadoop.fs.Path;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.junit.AfterClass;
@@ -50,7 +51,6 @@
import org.junit.rules.TemporaryFolder;
import java.io.File;
-import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import scala.concurrent.duration.FiniteDuration;
@@ -110,29 +110,50 @@ public void testMultipleAMKill() throws Exception {
final int numberKillingAttempts = numberApplicationAttempts - 1;
String confDirPath =
System.getenv(ConfigConstants.ENV_FLINK_CONF_DIR);
final Configuration configuration =
GlobalConfiguration.loadConfiguration();
- TestingYarnClusterDescriptor flinkYarnClient = new
TestingYarnClusterDescriptor(
- configuration,
- getYarnConfiguration(),
- confDirPath,
- getYarnClient(),
- true);
- Assert.assertNotNull("unable to get yarn client",
flinkYarnClient);
- flinkYarnClient.setLocalJarPath(new
Path(flinkUberjar.getAbsolutePath()));
-
flinkYarnClient.addShipFiles(Arrays.asList(flinkLibFolder.listFiles()));
+ configuration.setString(YarnConfigOptions.FLINK_JAR,
flinkUberjar.getAbsolutePath());
-
[jira] [Commented] (FLINK-10050) Support 'allowedLateness' in CoGroupedStreams
[ https://issues.apache.org/jira/browse/FLINK-10050?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620456#comment-16620456 ] ASF GitHub Bot commented on FLINK-10050: kl0u commented on issue #6646: [FLINK-10050] Support allowedLateness in CoGroupedStreams URL: https://github.com/apache/flink/pull/6646#issuecomment-422757135 I would recommend to relaunch that specific build and see if it happens again. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Support 'allowedLateness' in CoGroupedStreams > - > > Key: FLINK-10050 > URL: https://issues.apache.org/jira/browse/FLINK-10050 > Project: Flink > Issue Type: Improvement > Components: Streaming >Affects Versions: 1.5.1, 1.6.0 >Reporter: eugen yushin >Priority: Major > Labels: pull-request-available, ready-to-commit, windows > > WindowedStream has a support of 'allowedLateness' feature, while > CoGroupedStreams are not. At the mean time, WindowedStream is an inner part > of CoGroupedStreams and all main functionality (like evictor/trigger/...) is > simply delegated to WindowedStream. > There's no chance to operate with late arriving data from previous steps in > cogroups (and joins). Consider the following flow: > a. read data from source1 -> aggregate data with allowed lateness > b. read data from source2 -> aggregate data with allowed lateness > c. cogroup/join streams a and b, and compare aggregated values > Step c doesn't accept any late data from steps a/b due to lack of > `allowedLateness` API call in CoGroupedStreams.java. > Scope: add method `WithWindow.allowedLateness` to Java API > (flink-streaming-java) and extend scala API (flink-streaming-scala). -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[GitHub] kl0u commented on issue #6646: [FLINK-10050] Support allowedLateness in CoGroupedStreams
kl0u commented on issue #6646: [FLINK-10050] Support allowedLateness in CoGroupedStreams URL: https://github.com/apache/flink/pull/6646#issuecomment-422757135 I would recommend to relaunch that specific build and see if it happens again. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-10050) Support 'allowedLateness' in CoGroupedStreams
[ https://issues.apache.org/jira/browse/FLINK-10050?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620450#comment-16620450 ] ASF GitHub Bot commented on FLINK-10050: EugeneYushin commented on issue #6646: [FLINK-10050] Support allowedLateness in CoGroupedStreams URL: https://github.com/apache/flink/pull/6646#issuecomment-422754323 I see one build from Travis has been finished with error: ``` 10:15:13.087 [ERROR] Failed to execute goal on project flink-storm-examples_2.11: Could not resolve dependencies for project org.apache.flink:flink-storm-examples_2.11:jar:1.7-SNAPSHOT: Could not transfer artifact org.apache.storm:storm-starter:jar:1.0.0 from/to central (http://repo.maven.apache.org/maven2): GET request of: org/apache/storm/storm-starter/1.0.0/storm-starter-1.0.0.jar from central failed: Connection reset -> [Help 1] ``` Looks like it's not related to proposed changes. What is the best way to deal with such kind of failures during CI? Is there any way to rerun PR? This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Support 'allowedLateness' in CoGroupedStreams > - > > Key: FLINK-10050 > URL: https://issues.apache.org/jira/browse/FLINK-10050 > Project: Flink > Issue Type: Improvement > Components: Streaming >Affects Versions: 1.5.1, 1.6.0 >Reporter: eugen yushin >Priority: Major > Labels: pull-request-available, ready-to-commit, windows > > WindowedStream has a support of 'allowedLateness' feature, while > CoGroupedStreams are not. At the mean time, WindowedStream is an inner part > of CoGroupedStreams and all main functionality (like evictor/trigger/...) is > simply delegated to WindowedStream. > There's no chance to operate with late arriving data from previous steps in > cogroups (and joins). Consider the following flow: > a. read data from source1 -> aggregate data with allowed lateness > b. read data from source2 -> aggregate data with allowed lateness > c. cogroup/join streams a and b, and compare aggregated values > Step c doesn't accept any late data from steps a/b due to lack of > `allowedLateness` API call in CoGroupedStreams.java. > Scope: add method `WithWindow.allowedLateness` to Java API > (flink-streaming-java) and extend scala API (flink-streaming-scala). -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[GitHub] EugeneYushin commented on issue #6646: [FLINK-10050] Support allowedLateness in CoGroupedStreams
EugeneYushin commented on issue #6646: [FLINK-10050] Support allowedLateness in CoGroupedStreams URL: https://github.com/apache/flink/pull/6646#issuecomment-422754323 I see one build from Travis has been finished with error: ``` 10:15:13.087 [ERROR] Failed to execute goal on project flink-storm-examples_2.11: Could not resolve dependencies for project org.apache.flink:flink-storm-examples_2.11:jar:1.7-SNAPSHOT: Could not transfer artifact org.apache.storm:storm-starter:jar:1.0.0 from/to central (http://repo.maven.apache.org/maven2): GET request of: org/apache/storm/storm-starter/1.0.0/storm-starter-1.0.0.jar from central failed: Connection reset -> [Help 1] ``` Looks like it's not related to proposed changes. What is the best way to deal with such kind of failures during CI? Is there any way to rerun PR? This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-10355) The order of the column should start from 1.
[
https://issues.apache.org/jira/browse/FLINK-10355?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620439#comment-16620439
]
ASF GitHub Bot commented on FLINK-10355:
fhueske commented on a change in pull request #6713: [FLINK-10355][table]the
order of the columns start from 1
URL: https://github.com/apache/flink/pull/6713#discussion_r218750570
##
File path:
flink-java/src/main/java/org/apache/flink/api/java/io/RowCsvInputFormat.java
##
@@ -199,7 +199,7 @@ protected boolean parseRecord(Object[] holders, byte[]
bytes, int offset, int nu
// check if something went wrong
if (startPos < 0) {
throw new
ParseException(String.format("Unexpected parser position for column %1$s of row
'%2$s'",
- field, new String(bytes, offset,
numBytes)));
+ field+1, new String(bytes, offset,
numBytes)));
Review comment:
Same as above.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> The order of the column should start from 1.
>
>
> Key: FLINK-10355
> URL: https://issues.apache.org/jira/browse/FLINK-10355
> Project: Flink
> Issue Type: Improvement
> Components: Table API SQL
>Affects Versions: 1.6.0
>Reporter: lihongli
>Priority: Major
> Labels: easyfix, pull-request-available
> Attachments: B0C32FD9-47FE-4F63-921F-A9E49C0CB5CD.png
>
>
> When I register an external Table using a CsvTableSource.It throws an
> exception :"Parsing error for column 1".But I finally found that the second
> column is the error column.I think that the order of the column should start
> from 1.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-10355) The order of the column should start from 1.
[
https://issues.apache.org/jira/browse/FLINK-10355?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620438#comment-16620438
]
ASF GitHub Bot commented on FLINK-10355:
fhueske commented on a change in pull request #6713: [FLINK-10355][table]the
order of the columns start from 1
URL: https://github.com/apache/flink/pull/6713#discussion_r218750490
##
File path:
flink-java/src/main/java/org/apache/flink/api/java/io/RowCsvInputFormat.java
##
@@ -177,7 +177,7 @@ protected boolean parseRecord(Object[] holders, byte[]
bytes, int offset, int nu
// the error state EMPTY_COLUMN is
ignored
if (parser.getErrorState() !=
FieldParser.ParseErrorState.EMPTY_COLUMN) {
throw new
ParseException(String.format("Parsing error for column %1$s of row '%2$s'
originated by %3$s: %4$s.",
- field, new
String(bytes, offset, numBytes), parser.getClass().getSimpleName(),
parser.getErrorState()));
+ field+1, new
String(bytes, offset, numBytes), parser.getClass().getSimpleName(),
parser.getErrorState()));
Review comment:
This change does not pass the style check. Please add whitespaces around
`+`, i.e., -> `field + 1`.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> The order of the column should start from 1.
>
>
> Key: FLINK-10355
> URL: https://issues.apache.org/jira/browse/FLINK-10355
> Project: Flink
> Issue Type: Improvement
> Components: Table API SQL
>Affects Versions: 1.6.0
>Reporter: lihongli
>Priority: Major
> Labels: easyfix, pull-request-available
> Attachments: B0C32FD9-47FE-4F63-921F-A9E49C0CB5CD.png
>
>
> When I register an external Table using a CsvTableSource.It throws an
> exception :"Parsing error for column 1".But I finally found that the second
> column is the error column.I think that the order of the column should start
> from 1.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[GitHub] fhueske commented on a change in pull request #6713: [FLINK-10355][table]the order of the columns start from 1
fhueske commented on a change in pull request #6713: [FLINK-10355][table]the
order of the columns start from 1
URL: https://github.com/apache/flink/pull/6713#discussion_r218750490
##
File path:
flink-java/src/main/java/org/apache/flink/api/java/io/RowCsvInputFormat.java
##
@@ -177,7 +177,7 @@ protected boolean parseRecord(Object[] holders, byte[]
bytes, int offset, int nu
// the error state EMPTY_COLUMN is
ignored
if (parser.getErrorState() !=
FieldParser.ParseErrorState.EMPTY_COLUMN) {
throw new
ParseException(String.format("Parsing error for column %1$s of row '%2$s'
originated by %3$s: %4$s.",
- field, new
String(bytes, offset, numBytes), parser.getClass().getSimpleName(),
parser.getErrorState()));
+ field+1, new
String(bytes, offset, numBytes), parser.getClass().getSimpleName(),
parser.getErrorState()));
Review comment:
This change does not pass the style check. Please add whitespaces around
`+`, i.e., -> `field + 1`.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] fhueske commented on a change in pull request #6713: [FLINK-10355][table]the order of the columns start from 1
fhueske commented on a change in pull request #6713: [FLINK-10355][table]the
order of the columns start from 1
URL: https://github.com/apache/flink/pull/6713#discussion_r218750570
##
File path:
flink-java/src/main/java/org/apache/flink/api/java/io/RowCsvInputFormat.java
##
@@ -199,7 +199,7 @@ protected boolean parseRecord(Object[] holders, byte[]
bytes, int offset, int nu
// check if something went wrong
if (startPos < 0) {
throw new
ParseException(String.format("Unexpected parser position for column %1$s of row
'%2$s'",
- field, new String(bytes, offset,
numBytes)));
+ field+1, new String(bytes, offset,
numBytes)));
Review comment:
Same as above.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[jira] [Commented] (FLINK-10050) Support 'allowedLateness' in CoGroupedStreams
[ https://issues.apache.org/jira/browse/FLINK-10050?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620435#comment-16620435 ] ASF GitHub Bot commented on FLINK-10050: EugeneYushin commented on issue #6646: [FLINK-10050] Support allowedLateness in CoGroupedStreams URL: https://github.com/apache/flink/pull/6646#issuecomment-422746729 @kl0u I didn't want to ping you until Travis is finished. Thanks for quick turnaround This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Support 'allowedLateness' in CoGroupedStreams > - > > Key: FLINK-10050 > URL: https://issues.apache.org/jira/browse/FLINK-10050 > Project: Flink > Issue Type: Improvement > Components: Streaming >Affects Versions: 1.5.1, 1.6.0 >Reporter: eugen yushin >Priority: Major > Labels: pull-request-available, ready-to-commit, windows > > WindowedStream has a support of 'allowedLateness' feature, while > CoGroupedStreams are not. At the mean time, WindowedStream is an inner part > of CoGroupedStreams and all main functionality (like evictor/trigger/...) is > simply delegated to WindowedStream. > There's no chance to operate with late arriving data from previous steps in > cogroups (and joins). Consider the following flow: > a. read data from source1 -> aggregate data with allowed lateness > b. read data from source2 -> aggregate data with allowed lateness > c. cogroup/join streams a and b, and compare aggregated values > Step c doesn't accept any late data from steps a/b due to lack of > `allowedLateness` API call in CoGroupedStreams.java. > Scope: add method `WithWindow.allowedLateness` to Java API > (flink-streaming-java) and extend scala API (flink-streaming-scala). -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10331) Fix unnecessary flush requests to the network stack
[
https://issues.apache.org/jira/browse/FLINK-10331?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620434#comment-16620434
]
ASF GitHub Bot commented on FLINK-10331:
NicoK closed pull request #6692: [FLINK-10331][network] reduce unnecesary
flushing
URL: https://github.com/apache/flink/pull/6692
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java
index 41ee03db259..8630acee9a8 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java
@@ -485,7 +485,7 @@ private void initializeWithPartialRecord(NonSpanningWrapper
partial, int nextRec
}
else {
// collect in memory
- ensureBufferCapacity(numBytesChunk);
+ ensureBufferCapacity(nextRecordLength);
partial.segment.get(partial.position, buffer,
0, numBytesChunk);
}
@@ -502,8 +502,8 @@ private void addNextChunkFromMemorySegment(MemorySegment
segment, int offset, in
int segmentRemaining = numBytes;
// check where to go. if we have a partial length, we
need to complete it first
if (this.lengthBuffer.position() > 0) {
- int toPut =
Math.min(this.lengthBuffer.remaining(), numBytes);
- segment.get(offset, this.lengthBuffer, toPut);
+ int toPut =
Math.min(this.lengthBuffer.remaining(), segmentRemaining);
+ segment.get(segmentPosition, this.lengthBuffer,
toPut);
// did we complete the length?
if (this.lengthBuffer.hasRemaining()) {
return;
@@ -515,6 +515,8 @@ private void addNextChunkFromMemorySegment(MemorySegment
segment, int offset, in
segmentRemaining -= toPut;
if (this.recordLength >
THRESHOLD_FOR_SPILLING) {
this.spillingChannel =
createSpillingChannel();
+ } else {
+
ensureBufferCapacity(this.recordLength);
}
}
}
@@ -527,9 +529,7 @@ private void addNextChunkFromMemorySegment(MemorySegment
segment, int offset, in
// spill to file
ByteBuffer toWrite =
segment.wrap(segmentPosition, toCopy);
this.spillingChannel.write(toWrite);
- }
- else {
- ensureBufferCapacity(accumulatedRecordBytes +
toCopy);
+ } else {
segment.get(segmentPosition, buffer,
this.accumulatedRecordBytes, toCopy);
}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
index 305f1842911..6fb067ef8c3 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
@@ -104,9 +104,9 @@ public void commit() {
* @return number of written bytes.
*/
public int finish() {
- positionMarker.markFinished();
+ int writtenBytes = positionMarker.markFinished();
commit();
- return getWrittenBytes();
+ return writtenBytes;
}
public boolean isFinished() {
@@ -118,18 +118,10 @@ public boolean isFull() {
return positionMarker.getCached() == getMaxCapacity();
}
- public boolean isEmpty() {
- return positionMarker.getCached() == 0;
- }
-
public int getMaxCapacity() {
return memorySegment.size();
}
- private
[GitHub] EugeneYushin commented on issue #6646: [FLINK-10050] Support allowedLateness in CoGroupedStreams
EugeneYushin commented on issue #6646: [FLINK-10050] Support allowedLateness in CoGroupedStreams URL: https://github.com/apache/flink/pull/6646#issuecomment-422746729 @kl0u I didn't want to ping you until Travis is finished. Thanks for quick turnaround This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] NicoK closed pull request #6692: [FLINK-10331][network] reduce unnecesary flushing
NicoK closed pull request #6692: [FLINK-10331][network] reduce unnecesary
flushing
URL: https://github.com/apache/flink/pull/6692
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java
index 41ee03db259..8630acee9a8 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java
@@ -485,7 +485,7 @@ private void initializeWithPartialRecord(NonSpanningWrapper
partial, int nextRec
}
else {
// collect in memory
- ensureBufferCapacity(numBytesChunk);
+ ensureBufferCapacity(nextRecordLength);
partial.segment.get(partial.position, buffer,
0, numBytesChunk);
}
@@ -502,8 +502,8 @@ private void addNextChunkFromMemorySegment(MemorySegment
segment, int offset, in
int segmentRemaining = numBytes;
// check where to go. if we have a partial length, we
need to complete it first
if (this.lengthBuffer.position() > 0) {
- int toPut =
Math.min(this.lengthBuffer.remaining(), numBytes);
- segment.get(offset, this.lengthBuffer, toPut);
+ int toPut =
Math.min(this.lengthBuffer.remaining(), segmentRemaining);
+ segment.get(segmentPosition, this.lengthBuffer,
toPut);
// did we complete the length?
if (this.lengthBuffer.hasRemaining()) {
return;
@@ -515,6 +515,8 @@ private void addNextChunkFromMemorySegment(MemorySegment
segment, int offset, in
segmentRemaining -= toPut;
if (this.recordLength >
THRESHOLD_FOR_SPILLING) {
this.spillingChannel =
createSpillingChannel();
+ } else {
+
ensureBufferCapacity(this.recordLength);
}
}
}
@@ -527,9 +529,7 @@ private void addNextChunkFromMemorySegment(MemorySegment
segment, int offset, in
// spill to file
ByteBuffer toWrite =
segment.wrap(segmentPosition, toCopy);
this.spillingChannel.write(toWrite);
- }
- else {
- ensureBufferCapacity(accumulatedRecordBytes +
toCopy);
+ } else {
segment.get(segmentPosition, buffer,
this.accumulatedRecordBytes, toCopy);
}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
index 305f1842911..6fb067ef8c3 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
@@ -104,9 +104,9 @@ public void commit() {
* @return number of written bytes.
*/
public int finish() {
- positionMarker.markFinished();
+ int writtenBytes = positionMarker.markFinished();
commit();
- return getWrittenBytes();
+ return writtenBytes;
}
public boolean isFinished() {
@@ -118,18 +118,10 @@ public boolean isFull() {
return positionMarker.getCached() == getMaxCapacity();
}
- public boolean isEmpty() {
- return positionMarker.getCached() == 0;
- }
-
public int getMaxCapacity() {
return memorySegment.size();
}
- private int getWrittenBytes() {
- return positionMarker.getCached();
- }
-
/**
* Holds a reference to the current writer position. Negative values
indicate that writer ({@link BufferBuilder}
* has finished.
[jira] [Commented] (FLINK-9061) add entropy to s3 path for better scalability
[ https://issues.apache.org/jira/browse/FLINK-9061?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620432#comment-16620432 ] ASF GitHub Bot commented on FLINK-9061: --- StephanEwen commented on issue #6604: [FLINK-9061] Optionally add entropy to checkpoint paths better S3 scalability URL: https://github.com/apache/flink/pull/6604#issuecomment-422746118 Fixed for 1.6 in a58fa985c4944dd0cf39fa622ec0aa4b35f21f44 For master, I will open a new PR that is rebased on the latest commits... This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > add entropy to s3 path for better scalability > - > > Key: FLINK-9061 > URL: https://issues.apache.org/jira/browse/FLINK-9061 > Project: Flink > Issue Type: Bug > Components: FileSystem, State Backends, Checkpointing >Affects Versions: 1.5.0, 1.4.2 >Reporter: Jamie Grier >Assignee: Indrajit Roychoudhury >Priority: Critical > Labels: pull-request-available > > I think we need to modify the way we write checkpoints to S3 for high-scale > jobs (those with many total tasks). The issue is that we are writing all the > checkpoint data under a common key prefix. This is the worst case scenario > for S3 performance since the key is used as a partition key. > > In the worst case checkpoints fail with a 500 status code coming back from S3 > and an internal error type of TooBusyException. > > One possible solution would be to add a hook in the Flink filesystem code > that allows me to "rewrite" paths. For example say I have the checkpoint > directory set to: > > s3://bucket/flink/checkpoints > > I would hook that and rewrite that path to: > > s3://bucket/[HASH]/flink/checkpoints, where HASH is the hash of the original > path > > This would distribute the checkpoint write load around the S3 cluster evenly. > > For reference: > https://aws.amazon.com/premiumsupport/knowledge-center/s3-bucket-performance-improve/ > > Any other people hit this issue? Any other ideas for solutions? This is a > pretty serious problem for people trying to checkpoint to S3. > > -Jamie > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-9061) add entropy to s3 path for better scalability
[
https://issues.apache.org/jira/browse/FLINK-9061?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620433#comment-16620433
]
ASF GitHub Bot commented on FLINK-9061:
---
StephanEwen closed pull request #6604: [FLINK-9061] Optionally add entropy to
checkpoint paths better S3 scalability
URL: https://github.com/apache/flink/pull/6604
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java
b/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java
index d451109ddba..ba2113ae616 100644
--- a/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java
+++ b/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java
@@ -576,29 +576,8 @@ public boolean exists(final Path f) throws IOException {
/**
* Opens an FSDataOutputStream at the indicated Path.
*
-* This method is deprecated, because most of its parameters are
ignored by most file systems.
-* To control for example the replication factor and block size in the
Hadoop Distributed File system,
-* make sure that the respective Hadoop configuration file is either
linked from the Flink configuration,
-* or in the classpath of either Flink or the user code.
-*
-* @param f
-*the file name to open
-* @param overwrite
-*if a file with this name already exists, then if true,
-*the file will be overwritten, and if false an error will be
thrown.
-* @param bufferSize
-*the size of the buffer to be used.
-* @param replication
-*required block replication for the file.
-* @param blockSize
-*the size of the file blocks
-*
-* @throws IOException Thrown, if the stream could not be opened
because of an I/O, or because
-* a file already exists at that path and the write
mode indicates to not
-* overwrite the file.
-*
-* @deprecated Deprecated because not well supported across types of
file systems.
-* Control the behavior of specific file systems via
configurations instead.
+* @deprecated Deprecated in favor of {@link #create(Path,
WriteOptions)} which offers better extensibility
+* to options that are supported only by some filesystems
implementations.
*/
@Deprecated
public FSDataOutputStream create(
@@ -648,6 +627,25 @@ public FSDataOutputStream create(Path f, boolean
overwrite) throws IOException {
*/
public abstract FSDataOutputStream create(Path f, WriteMode
overwriteMode) throws IOException;
+ /**
+* Creates a new file at the given path and opens an FSDataOutputStream
to that new file.
+*
+* This method takes various options, some of which are not
supported by all file systems
+* (such as controlling block size).
+*
+* Implementation note: This method should be abstract, but is
currently not in order to not break
+* backwards compatibility of this class with earlier Flink versions.
+*
+* @param f The path for the new file.
+* @param options The options to parametrize the file and stream
creation.
+* @return The stream to the new file at the target path.
+*
+* @throws IOException Thrown if an error occurs while creating the
file or opening the stream.
+*/
+ public FSDataOutputStream create(Path f, WriteOptions options) throws
IOException {
+ return create(f, options.getOverwrite());
+ }
+
/**
* Renames the file/directory src to dst.
*
diff --git
a/flink-core/src/main/java/org/apache/flink/core/fs/WriteOptions.java
b/flink-core/src/main/java/org/apache/flink/core/fs/WriteOptions.java
new file mode 100644
index 000..70f4973dc14
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/core/fs/WriteOptions.java
@@ -0,0 +1,144 @@
+/*
+ * 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
[GitHub] StephanEwen commented on issue #6604: [FLINK-9061] Optionally add entropy to checkpoint paths better S3 scalability
StephanEwen commented on issue #6604: [FLINK-9061] Optionally add entropy to checkpoint paths better S3 scalability URL: https://github.com/apache/flink/pull/6604#issuecomment-422746118 Fixed for 1.6 in a58fa985c4944dd0cf39fa622ec0aa4b35f21f44 For master, I will open a new PR that is rebased on the latest commits... This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] StephanEwen closed pull request #6604: [FLINK-9061] Optionally add entropy to checkpoint paths better S3 scalability
StephanEwen closed pull request #6604: [FLINK-9061] Optionally add entropy to
checkpoint paths better S3 scalability
URL: https://github.com/apache/flink/pull/6604
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java
b/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java
index d451109ddba..ba2113ae616 100644
--- a/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java
+++ b/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java
@@ -576,29 +576,8 @@ public boolean exists(final Path f) throws IOException {
/**
* Opens an FSDataOutputStream at the indicated Path.
*
-* This method is deprecated, because most of its parameters are
ignored by most file systems.
-* To control for example the replication factor and block size in the
Hadoop Distributed File system,
-* make sure that the respective Hadoop configuration file is either
linked from the Flink configuration,
-* or in the classpath of either Flink or the user code.
-*
-* @param f
-*the file name to open
-* @param overwrite
-*if a file with this name already exists, then if true,
-*the file will be overwritten, and if false an error will be
thrown.
-* @param bufferSize
-*the size of the buffer to be used.
-* @param replication
-*required block replication for the file.
-* @param blockSize
-*the size of the file blocks
-*
-* @throws IOException Thrown, if the stream could not be opened
because of an I/O, or because
-* a file already exists at that path and the write
mode indicates to not
-* overwrite the file.
-*
-* @deprecated Deprecated because not well supported across types of
file systems.
-* Control the behavior of specific file systems via
configurations instead.
+* @deprecated Deprecated in favor of {@link #create(Path,
WriteOptions)} which offers better extensibility
+* to options that are supported only by some filesystems
implementations.
*/
@Deprecated
public FSDataOutputStream create(
@@ -648,6 +627,25 @@ public FSDataOutputStream create(Path f, boolean
overwrite) throws IOException {
*/
public abstract FSDataOutputStream create(Path f, WriteMode
overwriteMode) throws IOException;
+ /**
+* Creates a new file at the given path and opens an FSDataOutputStream
to that new file.
+*
+* This method takes various options, some of which are not
supported by all file systems
+* (such as controlling block size).
+*
+* Implementation note: This method should be abstract, but is
currently not in order to not break
+* backwards compatibility of this class with earlier Flink versions.
+*
+* @param f The path for the new file.
+* @param options The options to parametrize the file and stream
creation.
+* @return The stream to the new file at the target path.
+*
+* @throws IOException Thrown if an error occurs while creating the
file or opening the stream.
+*/
+ public FSDataOutputStream create(Path f, WriteOptions options) throws
IOException {
+ return create(f, options.getOverwrite());
+ }
+
/**
* Renames the file/directory src to dst.
*
diff --git
a/flink-core/src/main/java/org/apache/flink/core/fs/WriteOptions.java
b/flink-core/src/main/java/org/apache/flink/core/fs/WriteOptions.java
new file mode 100644
index 000..70f4973dc14
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/core/fs/WriteOptions.java
@@ -0,0 +1,144 @@
+/*
+ * 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
[jira] [Commented] (FLINK-8424) [Cassandra Connector] Update Cassandra version to last one
[ https://issues.apache.org/jira/browse/FLINK-8424?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16620412#comment-16620412 ] ASF GitHub Bot commented on FLINK-8424: --- bmeriaux commented on issue #6715: FLINK-8424: update cassandra and driver version to latest URL: https://github.com/apache/flink/pull/6715#issuecomment-422742511 i was a little lost on what part to keep and what to remove, is it fine now ? i saw both kinesis kinesis failed, i'm looking into it This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > [Cassandra Connector] Update Cassandra version to last one > -- > > Key: FLINK-8424 > URL: https://issues.apache.org/jira/browse/FLINK-8424 > Project: Flink > Issue Type: Improvement > Components: Cassandra Connector >Reporter: Joao Boto >Priority: Major > Labels: pull-request-available > > Cassandra connector are using a version outdated > This is to upgrade the cassandra version to something new > https://git1-us-west.apache.org/repos/asf?p=cassandra.git;a=blob_plain;f=CHANGES.txt;hb=refs/tags/cassandra-3.11.1 -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[GitHub] bmeriaux commented on issue #6715: FLINK-8424: update cassandra and driver version to latest
bmeriaux commented on issue #6715: FLINK-8424: update cassandra and driver version to latest URL: https://github.com/apache/flink/pull/6715#issuecomment-422742511 i was a little lost on what part to keep and what to remove, is it fine now ? i saw both kinesis kinesis failed, i'm looking into it This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
