[jira] [Commented] (FLINK-10696) Add APIs to ExternalCatalog for views and UDFs

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10696:


bowenli86 commented on issue #6970: [FLINK-10696][Table API & SQL]Add APIs to 
ExternalCatalog, CrudExternalCatalog and InMemoryCrudExternalCatalog for views 
and UDFs
URL: https://github.com/apache/flink/pull/6970#issuecomment-434570473
 
 
   @xuefuz @twalthr can you please take a look?


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 APIs to ExternalCatalog for views and UDFs
> --
>
> Key: FLINK-10696
> URL: https://issues.apache.org/jira/browse/FLINK-10696
> Project: Flink
>  Issue Type: Sub-task
>  Components: Table API  SQL
>Affects Versions: 1.6.1
>Reporter: Xuefu Zhang
>Assignee: Bowen Li
>Priority: Major
>  Labels: pull-request-available
>
> Currently there are APIs for tables only. However, views and UDFs are also 
> common objects in a catalog.
> This is required when we store Flink tables/views/UDFs in an external 
> persistent storage.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Updated] (FLINK-6944) Introduce DefaultTypeSerializerConfigSnapshot as a base implementation for serializer compatibility checks

[Tzu-Li (Gordon) Tai (JIRA)]

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

Tzu-Li (Gordon) Tai updated FLINK-6944:
---
Fix Version/s: (was: 1.7.0)

> Introduce DefaultTypeSerializerConfigSnapshot as a base implementation for 
> serializer compatibility checks
> --
>
> Key: FLINK-6944
> URL: https://issues.apache.org/jira/browse/FLINK-6944
> Project: Flink
>  Issue Type: Sub-task
>  Components: State Backends, Checkpointing, Type Serialization System
>Affects Versions: 1.3.0, 1.3.1
>Reporter: Tzu-Li (Gordon) Tai
>Priority: Critical
>
> FLINK-9377 proposes to remove writing serializers as part of checkpoint meta 
> info, and only write its configuration snapshot.
> Since then serializer config snapshots will be the single source of truth for 
> previous serializer schema, this JIRA proposes to follow up the change in 
> FLINK-9377 by having a base default implementation for letting 
> {{TypeSerializer.snapshotConfiguration}} that returns a 
> {{DefaultTypeSerializerConfigSnapshot}}.
> The {{DefaultTypeSerializerConfigSnapshot}} should wrap the serialVersionUID 
> of the serializer class, and the serializer class' classname. The latter two 
> will be used to check compatibility in the default implementation of 
> {{TypeSerializer.ensureCompatibility}}. Specifically, if classname / 
> serialVersionUID has changed, the default implementation of 
> {{TypeSerializer.ensureCompatibility}} should return 
> {{CompatibilityResult.requiresMigration}}.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Closed] (FLINK-9608) Add test base for TypeSerializer snapshotting / restore / migration to be used by all Flink serializers

[Tzu-Li (Gordon) Tai (JIRA)]

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

Tzu-Li (Gordon) Tai closed FLINK-9608.
--
   Resolution: Fixed
 Assignee: Igal Shilman  (was: Tzu-Li (Gordon) Tai)
Fix Version/s: (was: 1.7.0)

> Add test base for TypeSerializer snapshotting / restore / migration to be 
> used by all Flink serializers
> ---
>
> Key: FLINK-9608
> URL: https://issues.apache.org/jira/browse/FLINK-9608
> Project: Flink
>  Issue Type: Sub-task
>  Components: State Backends, Checkpointing, Tests, Type Serialization 
> System
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Igal Shilman
>Priority: Major
>
> The current {{TypeSerializerTestBase}} does not cover several aspects related 
> to upgrading / snapshotting and restoring Flink's serializers. That includes:
> - A serializer can be snapshotted and then restored without issues. The 
> restored serializer should be able to correctly read data written by the 
> snapshotted serializer.
> - A serializer's snapshotted configuration can be migrated from previous 
> Flink versions.
> - A serializer works with different kinds of state backends.
> This new proposed test base should go into {{flink-tests}}, with a coverage 
> check that all Flink's {{TypeSerializers}} have a corresponding subclass for 
> the test base.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Assigned] (FLINK-6944) Introduce DefaultTypeSerializerConfigSnapshot as a base implementation for serializer compatibility checks

[Tzu-Li (Gordon) Tai (JIRA)]

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

Tzu-Li (Gordon) Tai reassigned FLINK-6944:
--

Assignee: (was: Tzu-Li (Gordon) Tai)

> Introduce DefaultTypeSerializerConfigSnapshot as a base implementation for 
> serializer compatibility checks
> --
>
> Key: FLINK-6944
> URL: https://issues.apache.org/jira/browse/FLINK-6944
> Project: Flink
>  Issue Type: Sub-task
>  Components: State Backends, Checkpointing, Type Serialization System
>Affects Versions: 1.3.0, 1.3.1
>Reporter: Tzu-Li (Gordon) Tai
>Priority: Critical
>
> FLINK-9377 proposes to remove writing serializers as part of checkpoint meta 
> info, and only write its configuration snapshot.
> Since then serializer config snapshots will be the single source of truth for 
> previous serializer schema, this JIRA proposes to follow up the change in 
> FLINK-9377 by having a base default implementation for letting 
> {{TypeSerializer.snapshotConfiguration}} that returns a 
> {{DefaultTypeSerializerConfigSnapshot}}.
> The {{DefaultTypeSerializerConfigSnapshot}} should wrap the serialVersionUID 
> of the serializer class, and the serializer class' classname. The latter two 
> will be used to check compatibility in the default implementation of 
> {{TypeSerializer.ensureCompatibility}}. Specifically, if classname / 
> serialVersionUID has changed, the default implementation of 
> {{TypeSerializer.ensureCompatibility}} should return 
> {{CompatibilityResult.requiresMigration}}.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Closed] (FLINK-9808) Implement state conversion procedure in state backends

[Tzu-Li (Gordon) Tai (JIRA)]

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

Tzu-Li (Gordon) Tai closed FLINK-9808.
--
Resolution: Fixed

Merged for 1.7.0: 70f4141369850dcc16efe3ddfe71618ea8cda304

> Implement state conversion procedure in state backends
> --
>
> Key: FLINK-9808
> URL: https://issues.apache.org/jira/browse/FLINK-9808
> Project: Flink
>  Issue Type: Sub-task
>  Components: State Backends, Checkpointing
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Aljoscha Krettek
>Priority: Critical
>  Labels: pull-request-available
> Fix For: 1.7.0
>
>
> With FLINK-9377 in place and that config snapshots serve as the single source 
> of truth for recreating restore serializers, the next step would be to 
> utilize this when performing a full-pass state conversion (i.e., read with 
> old / restore serializer, write with new serializer).
> For Flink's heap-based backends, it can be seen that state conversion 
> inherently happens, since all state is always deserialized after restore with 
> the restore serializer, and written with the new serializer on snapshots.
> For the RocksDB state backend, since state is lazily deserialized, state 
> conversion needs to happen for per-registered state on their first access if 
> the registered new serializer has a different serialization schema than the 
> previous serializer.
> This task should consist of three parts:
> 1. Allow {{CompatibilityResult}} to correctly distinguish between whether the 
> new serializer's schema is a) compatible with the serializer as it is, b) 
> compatible after the serializer has been reconfigured, or c) incompatible.
> 2. Introduce state conversion procedures in the RocksDB state backend. This 
> should occur on the first state access.
> 3. Make sure that all other backends no longer do redundant serializer 
> compatibility checks. That is not required because those backends always 
> perform full-pass state conversions.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[GitHub] bowenli86 commented on issue #6970: [FLINK-10696][Table API & SQL]Add APIs to ExternalCatalog, CrudExternalCatalog and InMemoryCrudExternalCatalog for views and UDFs

[GitBox]

bowenli86 commented on issue #6970: [FLINK-10696][Table API & SQL]Add APIs to 
ExternalCatalog, CrudExternalCatalog and InMemoryCrudExternalCatalog for views 
and UDFs
URL: https://github.com/apache/flink/pull/6970#issuecomment-434570473
 
 
   @xuefuz @twalthr can you please take a look?


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] [Closed] (FLINK-10605) Upgrade AvroSerializer snapshot to implement new TypeSerializerSnapshot interface

[Tzu-Li (Gordon) Tai (JIRA)]

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

Tzu-Li (Gordon) Tai closed FLINK-10605.
---
Resolution: Fixed

Merged for 1.7.0: b0b10edca7cf776defb83a739c9db2b8b55e5f70

> Upgrade AvroSerializer snapshot to implement new TypeSerializerSnapshot 
> interface
> -
>
> Key: FLINK-10605
> URL: https://issues.apache.org/jira/browse/FLINK-10605
> Project: Flink
>  Issue Type: Sub-task
>  Components: Type Serialization System
>Reporter: Igal Shilman
>Assignee: Igal Shilman
>Priority: Major
>  Labels: pull-request-available
> Fix For: 1.7.0
>
>
> This issue introduces a new AvroSerializerSnapshot implementation that 
> conforms to the new TypeSerializerSnapshot API.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Closed] (FLINK-10679) Let TypeSerializerSchemaCompatibility.resolveSchemaCompatibility() be the entry point for compatibility checks in framework code

[Tzu-Li (Gordon) Tai (JIRA)]

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

Tzu-Li (Gordon) Tai closed FLINK-10679.
---
   Resolution: Fixed
Fix Version/s: 1.7.0

Merged for 1.7.0: a567a1ef628eadad21e11864ec328481cd6d7898

> Let TypeSerializerSchemaCompatibility.resolveSchemaCompatibility() be the 
> entry point for compatibility checks in framework code
> 
>
> Key: FLINK-10679
> URL: https://issues.apache.org/jira/browse/FLINK-10679
> Project: Flink
>  Issue Type: Sub-task
>  Components: State Backends, Checkpointing, Type Serialization System
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Stephan Ewen
>Priority: Major
>  Labels: pull-request-available
> Fix For: 1.7.0
>
>
> Currently, the state backend framework code still is exposed the now 
> deprecated {{CompatibilityResult}} and relevant classes.
> Instead, all compatibility checks should go through the new 
> {{TypeSerializerSchemaCompatibility#resolveSchemaCompatibility}} method, and 
> allow framework code to check against the more powerful 
> {{TypeSerializerSchemaCompatibility}} for incompatibility / migration 
> requirements.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Closed] (FLINK-10708) Implement a SerializerMigrationTestBase

[Tzu-Li (Gordon) Tai (JIRA)]

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

Tzu-Li (Gordon) Tai closed FLINK-10708.
---
   Resolution: Fixed
Fix Version/s: 1.7.0

Merged for 1.7.0:
a07cb5d9a3e4cb6c1ac0e6674b8d06ad33ec2148

> Implement a SerializerMigrationTestBase
> ---
>
> Key: FLINK-10708
> URL: https://issues.apache.org/jira/browse/FLINK-10708
> Project: Flink
>  Issue Type: Sub-task
>  Components: State Backends, Checkpointing, Type Serialization System
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Igal Shilman
>Priority: Major
> Fix For: 1.7.0
>
>
> The {{SerializerMigrationTestBase}} should cover the following commonly for 
> each upgraded serializer and snapshot:
> 1. Verify that we can still load the snapshot, and currently instantiate a 
> restore serializer from it.
> 2. Verify that the restored serializer is capable of reading data written by 
> the prior snapshotted serializer.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Closed] (FLINK-10709) Remove NS generic parameter from TypeSerializerSchemaCompatibility

[Tzu-Li (Gordon) Tai (JIRA)]

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

Tzu-Li (Gordon) Tai closed FLINK-10709.
---
   Resolution: Fixed
Fix Version/s: 1.7.0

Merged for 1.7.0:
17f311aadacab2f6b0a19de552017dbfeb0681e1

> Remove NS generic parameter from TypeSerializerSchemaCompatibility
> --
>
> Key: FLINK-10709
> URL: https://issues.apache.org/jira/browse/FLINK-10709
> Project: Flink
>  Issue Type: Sub-task
>  Components: Type Serialization System
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Stephan Ewen
>Priority: Major
> Fix For: 1.7.0
>
>
> It actually isn't required that the {{TypeSerializerSchemaCompatibility}} has 
> the {{NS extends TypeSerializer}} generic.
> That was previously added as a future-proof for when we eventually want to 
> add the {{NS getReconfiguredSerializer}} method on the compatibility result. 
> However, it would actually be beneficial if the reconfigured serializer isn't 
> strictly required to be of the same serializer type.
> Moreover, the NS generic makes serializer snapshot implementations more 
> complicated.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Closed] (FLINK-10716) Upgrade ListSerializer / ArrayListSerializer for state evolution

[Tzu-Li (Gordon) Tai (JIRA)]

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

Tzu-Li (Gordon) Tai closed FLINK-10716.
---
   Resolution: Fixed
Fix Version/s: 1.7.0

Merged for 1.7.0: bcf14d5434ea16b1f17eb4acf67721bcad498a24

> Upgrade ListSerializer / ArrayListSerializer for state evolution
> 
>
> Key: FLINK-10716
> URL: https://issues.apache.org/jira/browse/FLINK-10716
> Project: Flink
>  Issue Type: Sub-task
>  Components: Type Serialization System
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Tzu-Li (Gordon) Tai
>Priority: Major
> Fix For: 1.7.0
>
>
> Besides upgrading their snapshots, we should also make sure we have a 
> corresponding {{ListSerializerMigrationTest}} and 
> {{ArrayListSerializerMigrationTest}} to safe guard the upgrade.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Closed] (FLINK-10710) Implement new CompositeSerializerSnapshot

[Tzu-Li (Gordon) Tai (JIRA)]

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

Tzu-Li (Gordon) Tai closed FLINK-10710.
---
   Resolution: Fixed
Fix Version/s: 1.7.0

Merged for 1.7.0:
f86f72da6826f61d2c7e197c889ac4b85918efbf

> Implement new CompositeSerializerSnapshot
> -
>
> Key: FLINK-10710
> URL: https://issues.apache.org/jira/browse/FLINK-10710
> Project: Flink
>  Issue Type: Sub-task
>  Components: Type Serialization System
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Stephan Ewen
>Priority: Major
> Fix For: 1.7.0
>
>
> This ticket tracks the introduction of a new {{CompositeSerializerSnapshot}} 
> for the effort of upgrading our composite serializers and their snapshots to 
> their interfaces.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Closed] (FLINK-10717) Introduce SimpleSerializerSnapshot as replacement for ParameterlessTypeSerializerConfig

[Tzu-Li (Gordon) Tai (JIRA)]

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

Tzu-Li (Gordon) Tai closed FLINK-10717.
---
   Resolution: Fixed
Fix Version/s: 1.7.0

Merged for 1.7.0: 99c8272de8d3ac1b766b9c963e6d191a7906acd8

> Introduce SimpleSerializerSnapshot as replacement for 
> ParameterlessTypeSerializerConfig
> ---
>
> Key: FLINK-10717
> URL: https://issues.apache.org/jira/browse/FLINK-10717
> Project: Flink
>  Issue Type: Sub-task
>  Components: Type Serialization System
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Stephan Ewen
>Priority: Major
> Fix For: 1.7.0
>
>




--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Closed] (FLINK-10723) Upgrade MapSerializer for state evolution

[Tzu-Li (Gordon) Tai (JIRA)]

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

Tzu-Li (Gordon) Tai closed FLINK-10723.
---

> Upgrade MapSerializer for state evolution
> -
>
> Key: FLINK-10723
> URL: https://issues.apache.org/jira/browse/FLINK-10723
> Project: Flink
>  Issue Type: Sub-task
>  Components: Type Serialization System
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Tzu-Li (Gordon) Tai
>Priority: Major
> Fix For: 1.7.0
>
>
> Besides upgrading their snapshots, we should also make sure we have a 
> corresponding {{MapSerializerSnapshotMigrationTest}} to safe guard the 
> upgrade.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Resolved] (FLINK-10723) Upgrade MapSerializer for state evolution

[Tzu-Li (Gordon) Tai (JIRA)]

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

Tzu-Li (Gordon) Tai resolved FLINK-10723.
-
   Resolution: Fixed
Fix Version/s: 1.7.0

Merged for 1.7.0: ea0aa5162e459f5915f3c6939d7fbd9a880e8122

> Upgrade MapSerializer for state evolution
> -
>
> Key: FLINK-10723
> URL: https://issues.apache.org/jira/browse/FLINK-10723
> Project: Flink
>  Issue Type: Sub-task
>  Components: Type Serialization System
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Tzu-Li (Gordon) Tai
>Priority: Major
> Fix For: 1.7.0
>
>
> Besides upgrading their snapshots, we should also make sure we have a 
> corresponding {{MapSerializerSnapshotMigrationTest}} to safe guard the 
> upgrade.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Updated] (FLINK-9376) Allow upgrading to incompatible state serializers (state schema evolution)

[Tzu-Li (Gordon) Tai (JIRA)]

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

Tzu-Li (Gordon) Tai updated FLINK-9376:
---
Fix Version/s: (was: 1.7.0)

> Allow upgrading to incompatible state serializers (state schema evolution)
> --
>
> Key: FLINK-9376
> URL: https://issues.apache.org/jira/browse/FLINK-9376
> Project: Flink
>  Issue Type: New Feature
>  Components: State Backends, Checkpointing, Type Serialization System
>Reporter: Tzu-Li (Gordon) Tai
>Priority: Critical
>  Labels: pull-request-available
>
> Currently, users have access to upgrade state serializers on the restore run 
> of a stateful job, as long as the upgraded new serializer remains backwards 
> compatible with all previous written data in the savepoint (i.e. it can read 
> all previous and current schema of serialized state objects).
> What is still lacking is the ability to upgrade to incompatible serializers. 
> Upon being registered an incompatible serializer for existing restored state, 
> that state needs to go through the process of -
>  1. read serialized state with the previous serializer
>  2. passing each deserialized state object through a “migration map 
> function”, and
>  3. writing back the state with the new serializer
> The availability of this process should be strictly limited to state 
> registrations that occur before the actual processing begins (e.g. in the 
> {{open}} or {{initializeState}} methods), so that we avoid performing these 
> operations during processing.
> How this procedure actually occurs, differs across different types of state 
> backends.
> For example, for state backends that eagerly deserialize / lazily serialize 
> state (e.g. {{HeapStateBackend}}), the job execution itself can be seen as a 
> "migration"; everything is deserialized to state objects on restore, and is 
> only serialized again, with the new serializer, on checkpoints.
> Therefore, for these state backends, the above process is irrelevant.
> On the other hand, for state backends that lazily deserialize / eagerly 
> serialize state (e.g. {{RocksDBStateBackend}}), the state evolution process 
> needs to happen for every state with a newly registered incompatible 
> serializer.
> Procedure 2. will allow even state type migrations, but that is out-of-scope 
> of this JIRA.
>  This ticket focuses only on procedures 1. and 3., where we try to enable 
> schema evolution without state type changes.
> This is an umbrella JIRA ticket that overlooks this feature, including a few 
> preliminary tasks that work towards enabling it.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-10556) Integration with Apache Hive

[Bowen Li (JIRA)]

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

Bowen Li commented on FLINK-10556:
--

sorry that I put the wrong jira number in one of my PRs and got linked here...

Thanks, Xuefu! The design doc looks very good to me! 

One small thing caught my eyes is external catalog's compatibility towards 
current YAML config file used by SQL Client, as mentioned in section 
`Additional Notes`. The parsing logic of YAML config file currently is on a 
different code path than catalog's, and the diverge will create maintenance 
issue. Providing a migration utility, as suggested in design doc, is one way to 
solve it, while we still need to maintain the utility tool. In the long run, 
I'd suggest unifying the code paths of YAML config file and catalog by 
developing something like a {{FlinkYamlCatalog}}. It will make both maintenance 
and migration a lot easier. What do you think?

> Integration with Apache Hive
> 
>
> Key: FLINK-10556
> URL: https://issues.apache.org/jira/browse/FLINK-10556
> Project: Flink
>  Issue Type: New Feature
>  Components: Batch Connectors and Input/Output Formats, SQL Client, 
> Table API  SQL
>Affects Versions: 1.6.0
>Reporter: Xuefu Zhang
>Assignee: Xuefu Zhang
>Priority: Major
>  Labels: pull-request-available
> Attachments: Flink-Hive Metastore Connectivity Design.pdf, Proposal_ 
> Integrate Flink with Hive Ecosystem.pdf
>
>
> This is an umbrella JIRA tracking all enhancement and issues related to 
> integrating Flink with Hive ecosystem. This is an outcome of a discussion in 
> the community, and thanks go to everyone that provided feedback and interest.
> Specifically, we'd like to see the following features and capabilities 
> immediately in Flink:
> # Metadata interoperability
> # Data interoperability
> # Data type compatibility
> # Hive UDF support
> # DDL/DML/Query language compatibility
> For a longer term, we'd also like to add or improve:
> # Compatible SQL service, client tools, JDBC/ODBC drivers
> # Better task failure tolerance and task scheduling
> # Support other user customizations in Hive (storage handlers, serdes, etc).
> I will provide more details regarding the proposal in a doc shortly. Design 
> doc, if deemed necessary, will be provided in each related sub tasks under 
> this JIRA.
> Feedback and contributions are greatly welcome!



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-10696) Add APIs to ExternalCatalog for views and UDFs

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10696:


bowenli86 opened a new pull request #6970: [FLINK-10696][Table API & SQL]Add 
APIs to ExternalCatalog, CrudExternalCatalog and InMemoryCrudExternalCatalog 
for views and UDFs
URL: https://github.com/apache/flink/pull/6970
 
 
   ## What is the purpose of the change
   
   Currently Flink's external catalog have APIs for tables only. However, views 
and UDFs are also common objects in a catalog.
   
   Adding initial APIs and in-memory implementations for views and UDFs to 
external catalog. These APIs are  required when we store Flink views and UDFs 
in an external persistent storage. These APIs will evolve as we make progress 
in Flink-Hive integration.
   
   ## Brief change log
   
   - added initial APIs for views and UDFs in `ExternalCatalog` and 
`CrudExternalCatalog`
   - added in-memory implementations  in `InMemoryCrudExternalCatalog`
   - added relevant tests
   
   ## Verifying this change
   
   This change added tests and can be verified as follows:
   
 - Added unit tests in `InMemoryExternalCatalogTest`
   
   ## Does this pull request potentially affect one of the following parts:
   
   none
   
   ## Documentation
   
 - Does this pull request introduce a new feature? (yes )
 - If yes, how is the feature documented? (JavaDocs)
   


This is an automated message from the Apache Git Service.
To respond to the message, please log on 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 APIs to ExternalCatalog for views and UDFs
> --
>
> Key: FLINK-10696
> URL: https://issues.apache.org/jira/browse/FLINK-10696
> Project: Flink
>  Issue Type: Sub-task
>  Components: Table API  SQL
>Affects Versions: 1.6.1
>Reporter: Xuefu Zhang
>Assignee: Bowen Li
>Priority: Major
>  Labels: pull-request-available
>
> Currently there are APIs for tables only. However, views and UDFs are also 
> common objects in a catalog.
> This is required when we store Flink tables/views/UDFs in an external 
> persistent storage.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[GitHub] bowenli86 commented on issue #6969: [Table API & SQL]Add APIs to ExternalCatalog, CrudExternalCatalog and InMemoryCrudExternalCatalog for views and UDFs

[GitBox]

bowenli86 commented on issue #6969: [Table API & SQL]Add APIs to 
ExternalCatalog, CrudExternalCatalog and InMemoryCrudExternalCatalog for views 
and UDFs
URL: https://github.com/apache/flink/pull/6969#issuecomment-434557089
 
 
   put the wrong jira ticket number


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-10696) Add APIs to ExternalCatalog for views and UDFs

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot updated FLINK-10696:
---
Labels: pull-request-available  (was: )

> Add APIs to ExternalCatalog for views and UDFs
> --
>
> Key: FLINK-10696
> URL: https://issues.apache.org/jira/browse/FLINK-10696
> Project: Flink
>  Issue Type: Sub-task
>  Components: Table API  SQL
>Affects Versions: 1.6.1
>Reporter: Xuefu Zhang
>Assignee: Bowen Li
>Priority: Major
>  Labels: pull-request-available
>
> Currently there are APIs for tables only. However, views and UDFs are also 
> common objects in a catalog.
> This is required when we store Flink tables/views/UDFs in an external 
> persistent storage.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[GitHub] bowenli86 opened a new pull request #6970: [FLINK-10696][Table API & SQL]Add APIs to ExternalCatalog, CrudExternalCatalog and InMemoryCrudExternalCatalog for views and UDFs

[GitBox]

bowenli86 opened a new pull request #6970: [FLINK-10696][Table API & SQL]Add 
APIs to ExternalCatalog, CrudExternalCatalog and InMemoryCrudExternalCatalog 
for views and UDFs
URL: https://github.com/apache/flink/pull/6970
 
 
   ## What is the purpose of the change
   
   Currently Flink's external catalog have APIs for tables only. However, views 
and UDFs are also common objects in a catalog.
   
   Adding initial APIs and in-memory implementations for views and UDFs to 
external catalog. These APIs are  required when we store Flink views and UDFs 
in an external persistent storage. These APIs will evolve as we make progress 
in Flink-Hive integration.
   
   ## Brief change log
   
   - added initial APIs for views and UDFs in `ExternalCatalog` and 
`CrudExternalCatalog`
   - added in-memory implementations  in `InMemoryCrudExternalCatalog`
   - added relevant tests
   
   ## Verifying this change
   
   This change added tests and can be verified as follows:
   
 - Added unit tests in `InMemoryExternalCatalogTest`
   
   ## Does this pull request potentially affect one of the following parts:
   
   none
   
   ## Documentation
   
 - Does this pull request introduce a new feature? (yes )
 - If yes, how is the feature documented? (JavaDocs)
   


This is an automated message from the Apache Git Service.
To respond to the message, please log on 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-10556) Integration with Apache Hive

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10556:


bowenli86 closed pull request #6969: [FLINK-10556][Table API & SQL]Add APIs to 
ExternalCatalog, CrudExternalCatalog and InMemoryCrudExternalCatalog for views 
and UDFs
URL: https://github.com/apache/flink/pull/6969
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/exceptions.scala
 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/exceptions.scala
index 7fc7de50e07..0f1c80df544 100644
--- 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/exceptions.scala
+++ 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/exceptions.scala
@@ -107,6 +107,70 @@ case class CatalogAlreadyExistException(
   def this(catalog: String) = this(catalog, null)
 }
 
+/**
+  * Exception for operation on a nonexistent view
+  *
+  * @param catalog catalog name
+  * @param view view name
+  * @param cause the cause
+  */
+case class ViewNotExistException(
+  catalog: String,
+  view: String,
+  cause: Throwable)
+  extends RuntimeException(s"View $view does not exist.", cause) {
+
+  def this(catalog: String, view: String) = this(catalog, view, null)
+}
+
+/**
+  * Exception for adding an already existent view
+  *
+  * @param catalog catalog name
+  * @param view view name
+  * @param cause the cause
+  */
+case class ViewAlreadyExistException(
+  catalog: String,
+  view: String,
+  cause: Throwable)
+  extends RuntimeException(s"View $view already exists.", cause) {
+
+  def this(catalog: String, view: String) = this(catalog, view, null)
+}
+
+/**
+  * Exception for operation on a nonexistent function
+  *
+  * @param catalog catalog name
+  * @param function function name
+  * @param cause the cause
+  */
+case class FunctionNotExistException(
+  catalog: String,
+  function: String,
+  cause: Throwable)
+  extends RuntimeException(s"Function $function does not exist.", cause) {
+
+  def this(catalog: String, function: String) = this(catalog, function, null)
+}
+
+/**
+  * Exception for adding an already existent function
+  *
+  * @param catalog catalog name
+  * @param function function name
+  * @param cause the cause
+  */
+case class FunctionAlreadyExistException(
+  catalog: String,
+  function: String,
+  cause: Throwable)
+  extends RuntimeException(s"Function $function already exists.", cause) {
+
+  def this(catalog: String, function: String) = this(catalog, function, null)
+}
+
 /**
   * Exception for not finding a [[TableFactory]] for the given properties.
   *
diff --git 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/CrudExternalCatalog.scala
 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/CrudExternalCatalog.scala
index 4db9497a712..02bdbed21fc 100644
--- 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/CrudExternalCatalog.scala
+++ 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/CrudExternalCatalog.scala
@@ -19,9 +19,11 @@
 package org.apache.flink.table.catalog
 
 import org.apache.flink.table.api._
+import org.apache.flink.table.functions.UserDefinedFunction
 
 /**
-  * The CrudExternalCatalog provides methods to create, drop, and alter 
(sub-)catalogs or tables.
+  * The CrudExternalCatalog provides methods to create, drop, and alter 
(sub-)catalogs, tables,
+  * views and UDFs.
   */
 trait CrudExternalCatalog extends ExternalCatalog {
 
@@ -103,4 +105,86 @@ trait CrudExternalCatalog extends ExternalCatalog {
   @throws[CatalogNotExistException]
   def alterSubCatalog(name: String, catalog: ExternalCatalog, 
ignoreIfNotExists: Boolean): Unit
 
+  /**
+* Adds a view to this catalog.
+*
+* @param viewName  The name of the view to add.
+* @param view  The view to add.
+* @param ignoreIfExists Flag to specify behavior if a view with the given 
name already exists:
+*   if set to false, throw an exception,
+*   if set to true, nothing happens.
+* @throws ViewAlreadyExistException thrown if view already exists and 
ignoreIfExists is false
+*/
+  @throws[ViewAlreadyExistException]
+  def createView(viewName: String, view: String, ignoreIfExists: Boolean): Unit
+
+  /**
+* Deletes a view from this catalog.
+*
+* @param viewName Name of the view to delete.
+* @param ignoreIfNotExists Flag to specify behavior if the view does not 
exist:

[GitHub] bowenli86 closed pull request #6969: [FLINK-10556][Table API & SQL]Add APIs to ExternalCatalog, CrudExternalCatalog and InMemoryCrudExternalCatalog for views and UDFs

[GitBox]

bowenli86 closed pull request #6969: [FLINK-10556][Table API & SQL]Add APIs to 
ExternalCatalog, CrudExternalCatalog and InMemoryCrudExternalCatalog for views 
and UDFs
URL: https://github.com/apache/flink/pull/6969
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/exceptions.scala
 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/exceptions.scala
index 7fc7de50e07..0f1c80df544 100644
--- 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/exceptions.scala
+++ 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/exceptions.scala
@@ -107,6 +107,70 @@ case class CatalogAlreadyExistException(
   def this(catalog: String) = this(catalog, null)
 }
 
+/**
+  * Exception for operation on a nonexistent view
+  *
+  * @param catalog catalog name
+  * @param view view name
+  * @param cause the cause
+  */
+case class ViewNotExistException(
+  catalog: String,
+  view: String,
+  cause: Throwable)
+  extends RuntimeException(s"View $view does not exist.", cause) {
+
+  def this(catalog: String, view: String) = this(catalog, view, null)
+}
+
+/**
+  * Exception for adding an already existent view
+  *
+  * @param catalog catalog name
+  * @param view view name
+  * @param cause the cause
+  */
+case class ViewAlreadyExistException(
+  catalog: String,
+  view: String,
+  cause: Throwable)
+  extends RuntimeException(s"View $view already exists.", cause) {
+
+  def this(catalog: String, view: String) = this(catalog, view, null)
+}
+
+/**
+  * Exception for operation on a nonexistent function
+  *
+  * @param catalog catalog name
+  * @param function function name
+  * @param cause the cause
+  */
+case class FunctionNotExistException(
+  catalog: String,
+  function: String,
+  cause: Throwable)
+  extends RuntimeException(s"Function $function does not exist.", cause) {
+
+  def this(catalog: String, function: String) = this(catalog, function, null)
+}
+
+/**
+  * Exception for adding an already existent function
+  *
+  * @param catalog catalog name
+  * @param function function name
+  * @param cause the cause
+  */
+case class FunctionAlreadyExistException(
+  catalog: String,
+  function: String,
+  cause: Throwable)
+  extends RuntimeException(s"Function $function already exists.", cause) {
+
+  def this(catalog: String, function: String) = this(catalog, function, null)
+}
+
 /**
   * Exception for not finding a [[TableFactory]] for the given properties.
   *
diff --git 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/CrudExternalCatalog.scala
 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/CrudExternalCatalog.scala
index 4db9497a712..02bdbed21fc 100644
--- 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/CrudExternalCatalog.scala
+++ 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/catalog/CrudExternalCatalog.scala
@@ -19,9 +19,11 @@
 package org.apache.flink.table.catalog
 
 import org.apache.flink.table.api._
+import org.apache.flink.table.functions.UserDefinedFunction
 
 /**
-  * The CrudExternalCatalog provides methods to create, drop, and alter 
(sub-)catalogs or tables.
+  * The CrudExternalCatalog provides methods to create, drop, and alter 
(sub-)catalogs, tables,
+  * views and UDFs.
   */
 trait CrudExternalCatalog extends ExternalCatalog {
 
@@ -103,4 +105,86 @@ trait CrudExternalCatalog extends ExternalCatalog {
   @throws[CatalogNotExistException]
   def alterSubCatalog(name: String, catalog: ExternalCatalog, 
ignoreIfNotExists: Boolean): Unit
 
+  /**
+* Adds a view to this catalog.
+*
+* @param viewName  The name of the view to add.
+* @param view  The view to add.
+* @param ignoreIfExists Flag to specify behavior if a view with the given 
name already exists:
+*   if set to false, throw an exception,
+*   if set to true, nothing happens.
+* @throws ViewAlreadyExistException thrown if view already exists and 
ignoreIfExists is false
+*/
+  @throws[ViewAlreadyExistException]
+  def createView(viewName: String, view: String, ignoreIfExists: Boolean): Unit
+
+  /**
+* Deletes a view from this catalog.
+*
+* @param viewName Name of the view to delete.
+* @param ignoreIfNotExists Flag to specify behavior if the view does not 
exist:
+*  if set to false, throw an exception,
+*  if set to true, nothing happens.
+* @throws ViewNotExistExceptionthrown if the view does not exist in 
the catalog
+*/
+  

[jira] [Updated] (FLINK-10556) Integration with Apache Hive

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot updated FLINK-10556:
---
Labels: pull-request-available  (was: )

> Integration with Apache Hive
> 
>
> Key: FLINK-10556
> URL: https://issues.apache.org/jira/browse/FLINK-10556
> Project: Flink
>  Issue Type: New Feature
>  Components: Batch Connectors and Input/Output Formats, SQL Client, 
> Table API  SQL
>Affects Versions: 1.6.0
>Reporter: Xuefu Zhang
>Assignee: Xuefu Zhang
>Priority: Major
>  Labels: pull-request-available
> Attachments: Flink-Hive Metastore Connectivity Design.pdf, Proposal_ 
> Integrate Flink with Hive Ecosystem.pdf
>
>
> This is an umbrella JIRA tracking all enhancement and issues related to 
> integrating Flink with Hive ecosystem. This is an outcome of a discussion in 
> the community, and thanks go to everyone that provided feedback and interest.
> Specifically, we'd like to see the following features and capabilities 
> immediately in Flink:
> # Metadata interoperability
> # Data interoperability
> # Data type compatibility
> # Hive UDF support
> # DDL/DML/Query language compatibility
> For a longer term, we'd also like to add or improve:
> # Compatible SQL service, client tools, JDBC/ODBC drivers
> # Better task failure tolerance and task scheduling
> # Support other user customizations in Hive (storage handlers, serdes, etc).
> I will provide more details regarding the proposal in a doc shortly. Design 
> doc, if deemed necessary, will be provided in each related sub tasks under 
> this JIRA.
> Feedback and contributions are greatly welcome!



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-10556) Integration with Apache Hive

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10556:


bowenli86 opened a new pull request #6969: [FLINK-10556][Table API & SQL]Add 
APIs to ExternalCatalog, CrudExternalCatalog and InMemoryCrudExternalCatalog 
for views and UDFs
URL: https://github.com/apache/flink/pull/6969
 
 
   ## What is the purpose of the change
   
   Currently Flink's external catalog have APIs for tables only. However, views 
and UDFs are also common objects in a catalog.
   
   Adding initial APIs and in-memory implementations for views and UDFs to 
external catalog. These APIs are  required when we store Flink views and UDFs 
in an external persistent storage. These APIs will evolve as we make progress 
in Flink-Hive integration.
   
   ## Brief change log
   
   - added initial APIs for views and UDFs in `ExternalCatalog` and 
`CrudExternalCatalog`
   - added in-memory implementations  in `InMemoryCrudExternalCatalog`
   - added relevant tests
   
   ## Verifying this change
   
   This change added tests and can be verified as follows:
   
 - Added unit tests in `InMemoryExternalCatalogTest`
   
   ## Does this pull request potentially affect one of the following parts:
   
   none
   
   ## Documentation
   
 - Does this pull request introduce a new feature? (yes )
 - If yes, how is the feature documented? (JavaDocs)
   


This is an automated message from the Apache Git Service.
To respond to the message, please log on 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


> Integration with Apache Hive
> 
>
> Key: FLINK-10556
> URL: https://issues.apache.org/jira/browse/FLINK-10556
> Project: Flink
>  Issue Type: New Feature
>  Components: Batch Connectors and Input/Output Formats, SQL Client, 
> Table API  SQL
>Affects Versions: 1.6.0
>Reporter: Xuefu Zhang
>Assignee: Xuefu Zhang
>Priority: Major
>  Labels: pull-request-available
> Attachments: Flink-Hive Metastore Connectivity Design.pdf, Proposal_ 
> Integrate Flink with Hive Ecosystem.pdf
>
>
> This is an umbrella JIRA tracking all enhancement and issues related to 
> integrating Flink with Hive ecosystem. This is an outcome of a discussion in 
> the community, and thanks go to everyone that provided feedback and interest.
> Specifically, we'd like to see the following features and capabilities 
> immediately in Flink:
> # Metadata interoperability
> # Data interoperability
> # Data type compatibility
> # Hive UDF support
> # DDL/DML/Query language compatibility
> For a longer term, we'd also like to add or improve:
> # Compatible SQL service, client tools, JDBC/ODBC drivers
> # Better task failure tolerance and task scheduling
> # Support other user customizations in Hive (storage handlers, serdes, etc).
> I will provide more details regarding the proposal in a doc shortly. Design 
> doc, if deemed necessary, will be provided in each related sub tasks under 
> this JIRA.
> Feedback and contributions are greatly welcome!



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[GitHub] bowenli86 opened a new pull request #6969: [FLINK-10556][Table API & SQL]Add APIs to ExternalCatalog, CrudExternalCatalog and InMemoryCrudExternalCatalog for views and UDFs

[GitBox]

bowenli86 opened a new pull request #6969: [FLINK-10556][Table API & SQL]Add 
APIs to ExternalCatalog, CrudExternalCatalog and InMemoryCrudExternalCatalog 
for views and UDFs
URL: https://github.com/apache/flink/pull/6969
 
 
   ## What is the purpose of the change
   
   Currently Flink's external catalog have APIs for tables only. However, views 
and UDFs are also common objects in a catalog.
   
   Adding initial APIs and in-memory implementations for views and UDFs to 
external catalog. These APIs are  required when we store Flink views and UDFs 
in an external persistent storage. These APIs will evolve as we make progress 
in Flink-Hive integration.
   
   ## Brief change log
   
   - added initial APIs for views and UDFs in `ExternalCatalog` and 
`CrudExternalCatalog`
   - added in-memory implementations  in `InMemoryCrudExternalCatalog`
   - added relevant tests
   
   ## Verifying this change
   
   This change added tests and can be verified as follows:
   
 - Added unit tests in `InMemoryExternalCatalogTest`
   
   ## Does this pull request potentially affect one of the following parts:
   
   none
   
   ## Documentation
   
 - Does this pull request introduce a new feature? (yes )
 - If yes, how is the feature documented? (JavaDocs)
   


This is an automated message from the Apache Git Service.
To respond to the message, please log on 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-5697) Add per-shard watermarks for FlinkKinesisConsumer

[Thomas Weise (JIRA)]

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

Thomas Weise commented on FLINK-5697:
-

The latest related discussion: 
[https://lists.apache.org/thread.html/2341ab54f5c9a4f43f123605d6b17e374c19587bfd72c22748e08652@%3Cdev.flink.apache.org%3E]

I'm planning to first bring over the source watermarking that we implemented a 
while ago at Lyft to the Flink Kinesis consumer. Following that, we are 
planning to contribute the emit queue pattern as described in Jamie's document 
and the state sharing mechanism. Those pieces will allow to back pressure the 
shard consumer threads based on the shared watermark information. 

 

> Add per-shard watermarks for FlinkKinesisConsumer
> -
>
> Key: FLINK-5697
> URL: https://issues.apache.org/jira/browse/FLINK-5697
> Project: Flink
>  Issue Type: New Feature
>  Components: Kinesis Connector, Streaming Connectors
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Thomas Weise
>Priority: Major
>
> It would be nice to let the Kinesis consumer be on-par in functionality with 
> the Kafka consumer, since they share very similar abstractions. Per-partition 
> / shard watermarks is something we can add also to the Kinesis consumer.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-7342) Add support for Regular Expression Operators (RLIKE)

[Rong Rong (JIRA)]

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

Rong Rong commented on FLINK-7342:
--

Seems like {{SIMILAR TO}} was already supported in Flink, should we close this 
ticket or is there any additional features we need to support using LIKE 
operator?

> Add support for Regular Expression Operators (RLIKE) 
> -
>
> Key: FLINK-7342
> URL: https://issues.apache.org/jira/browse/FLINK-7342
> Project: Flink
>  Issue Type: New Feature
>  Components: Table API  SQL
>Reporter: sunjincheng
>Assignee: sunjincheng
>Priority: Major
>
> Currently Regular Expression Operators (RLIKE)  are not supported. e.g.: 
> {code}
> select * from emps name rlike '^(K|G|W)' 
> {code}
> we'll get Error:
> {code}
> Error: Error while executing SQL "select * from emps name rlike '^(K|G|W)'": 
> parse failed: Encountered "rlike" at line 1, column 25.
> {code}
> So, after https://issues.apache.org/jira/browse/CALCITE-1918 fixed. I want 
> add RLIKE supported in flink (TableAPI and SQL).
> Reference:
> MySql: https://dev.mysql.com/doc/refman/5.7/en/regexp.html#operator_regexp



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Assigned] (FLINK-5697) Add per-shard watermarks for FlinkKinesisConsumer

[Thomas Weise (JIRA)]

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

Thomas Weise reassigned FLINK-5697:
---

Assignee: Thomas Weise

> Add per-shard watermarks for FlinkKinesisConsumer
> -
>
> Key: FLINK-5697
> URL: https://issues.apache.org/jira/browse/FLINK-5697
> Project: Flink
>  Issue Type: New Feature
>  Components: Kinesis Connector, Streaming Connectors
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Thomas Weise
>Priority: Major
>
> It would be nice to let the Kinesis consumer be on-par in functionality with 
> the Kafka consumer, since they share very similar abstractions. Per-partition 
> / shard watermarks is something we can add also to the Kinesis consumer.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Comment Edited] (FLINK-10231) Add a view SQL DDL

[winifredtang (JIRA)]

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

winifredtang edited comment on FLINK-10231 at 10/31/18 2:50 AM:


[~fhueske] Hello, could you give more details about it?  I try to achieve a 
view SQL DDL. I wonder if the TableEnvironment.registerTableInternal(name: 
String, table: AbstractTable) might be deprecated too. Thanks a lot.


was (Author: winipanda):
Fabian Hueske Hello, could you give more details about it?  I try to achieve a 
view SQL DDL. I wonder if the TableEnvironment.registerTableInternal(name: 
String, table: AbstractTable) might be deprecated too. Thanks a lot.

> Add a view SQL DDL
> --
>
> Key: FLINK-10231
> URL: https://issues.apache.org/jira/browse/FLINK-10231
> Project: Flink
>  Issue Type: Sub-task
>  Components: Table API  SQL
>Reporter: Timo Walther
>Assignee: winifredtang
>Priority: Major
>
> FLINK-10163 added initial view support for the SQL Client. However, for 
> supporting the [full definition of 
> views|https://cwiki.apache.org/confluence/display/Hive/LanguageManual+DDL#LanguageManualDDL-Create/Drop/AlterView]
>  (with schema, comments, etc.) we need to support native support for views in 
> the Table API.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Assigned] (FLINK-10729) Create a Hive a connector to access Hive data

[Zhenqiu Huang (JIRA)]

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

Zhenqiu Huang reassigned FLINK-10729:
-

Assignee: Zhenqiu Huang

> Create a Hive a connector to access Hive data
> -
>
> Key: FLINK-10729
> URL: https://issues.apache.org/jira/browse/FLINK-10729
> Project: Flink
>  Issue Type: Sub-task
>  Components: Table API  SQL
>Affects Versions: 1.6.2
>Reporter: Xuefu Zhang
>Assignee: Zhenqiu Huang
>Priority: Major
>
> As part of Flink-Hive integration effort, it's important for Flink to access 
> (read/write) Hive data, which is the responsibility of Hive connector. While 
> there is a HCatalog data connector in the code base, it's not complete (i.e. 
> missing all connector related classes such as validators, etc.). Further, 
> HCatalog interface has many limitations such as accessing a subset of Hive 
> data, supporting a subset of Hive data types, etc. In addition, it's not 
> actively maintained. In fact, it's now only a sub-project in Hive.
> Therefore, here we propose a complete connector set for Hive tables, not via 
> HCatalog, but via direct Hive interface. HCatalog connector will be 
> deprecated.
> Please note that connector on Hive metadata is already covered in other 
> JIRAs, as {{HiveExternalCatalog}}.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-10729) Create a Hive a connector to access Hive data

[Zhenqiu Huang (JIRA)]

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

Zhenqiu Huang commented on FLINK-10729:
---

Hi Xuefu, I am interested with this Task, Would you please assign it to me?

> Create a Hive a connector to access Hive data
> -
>
> Key: FLINK-10729
> URL: https://issues.apache.org/jira/browse/FLINK-10729
> Project: Flink
>  Issue Type: Sub-task
>  Components: Table API  SQL
>Affects Versions: 1.6.2
>Reporter: Xuefu Zhang
>Assignee: Zhenqiu Huang
>Priority: Major
>
> As part of Flink-Hive integration effort, it's important for Flink to access 
> (read/write) Hive data, which is the responsibility of Hive connector. While 
> there is a HCatalog data connector in the code base, it's not complete (i.e. 
> missing all connector related classes such as validators, etc.). Further, 
> HCatalog interface has many limitations such as accessing a subset of Hive 
> data, supporting a subset of Hive data types, etc. In addition, it's not 
> actively maintained. In fact, it's now only a sub-project in Hive.
> Therefore, here we propose a complete connector set for Hive tables, not via 
> HCatalog, but via direct Hive interface. HCatalog connector will be 
> deprecated.
> Please note that connector on Hive metadata is already covered in other 
> JIRAs, as {{HiveExternalCatalog}}.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Comment Edited] (FLINK-10231) Add a view SQL DDL

[winifredtang (JIRA)]

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

winifredtang edited comment on FLINK-10231 at 10/31/18 2:47 AM:


Fabian Hueske Hello, could you give more details about it?  I try to achieve a 
view SQL DDL. I wonder if the TableEnvironment.registerTableInternal(name: 
String, table: AbstractTable) might be deprecated too. Thanks a lot.


was (Author: winipanda):
[~fhueske] Hello, could you give more details about it?  I try to achieve a 
view SQL DDL. I wonder if it would better for me to use the 
TableEnvironment.registerTableInternal(name: String, table: AbstractTable) 
instead of the TableEnvironment.registerTable(name: String, table: Table). 
Thanks a lot.

> Add a view SQL DDL
> --
>
> Key: FLINK-10231
> URL: https://issues.apache.org/jira/browse/FLINK-10231
> Project: Flink
>  Issue Type: Sub-task
>  Components: Table API  SQL
>Reporter: Timo Walther
>Assignee: winifredtang
>Priority: Major
>
> FLINK-10163 added initial view support for the SQL Client. However, for 
> supporting the [full definition of 
> views|https://cwiki.apache.org/confluence/display/Hive/LanguageManual+DDL#LanguageManualDDL-Create/Drop/AlterView]
>  (with schema, comments, etc.) we need to support native support for views in 
> the Table API.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-10600) Provide End-to-end test cases for modern Kafka connectors

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10600:


yanghua edited a comment on issue #6924: [FLINK-10600] Provide End-to-end test 
cases for modern Kafka connectors
URL: https://github.com/apache/flink/pull/6924#issuecomment-434539461
 
 
   @pnowojski I can give you more details : 
   
   added `set -x` and `set -e`: 
   
   ```
   + stop_kafka_cluster
   + 
${FLINK_DIR}/flink-end-to-end-tests/test-scripts/temp-test-directory-01N/kafka_2.11-2.0.0/bin/kafka-server-stop.sh
   + 
${FLINK_DIR}/flink-end-to-end-tests/test-scripts/temp-test-directory-01N/kafka_2.11-2.0.0/bin/zookeeper-server-stop.sh
   ++ grep -i 'kafka\.Kafka'
   ++ jps -vl
   ++ grep java
   ++ grep -v grep
   ++ awk '{print $1}'
   + PIDS=55405
   + '[' '!' -z 55405 ']'
   + kill -s TERM 55405
   ++ jps -vl
   ++ grep java
   ++ grep -i QuorumPeerMain
   ++ grep -v grep
   ++ awk '{print $1}'
   + PIDS=
   
   [FAIL] 'Modern Kafka end-to-end test' failed after 0 minutes and 57 seconds! 
Test exited with exit code 1
   
   Stopping taskexecutor daemon (pid: 56949) on host ${MY_HOST}.
   Stopping standalonesession daemon (pid: 56518) on host ${MY_HOST}.
   No zookeeper daemon to stop on host ${MY_HOST}.
   ```
   
   But, just add `set -x`
   
   ```
   + stop_kafka_cluster
   + 
${FLINK_DIR}/flink-end-to-end-tests/test-scripts/temp-test-directory-21N/kafka_2.11-2.0.0/bin/kafka-server-stop.sh
   + 
${FLINK_DIR}/flink-end-to-end-tests/test-scripts/temp-test-directory-21N/kafka_2.11-2.0.0/bin/zookeeper-server-stop.sh
   ++ jps -vl
   ++ grep -i 'kafka\.Kafka'
   ++ grep java
   ++ grep -v grep
   ++ awk '{print $1}'
   + PIDS=62800
   + '[' '!' -z 62800 ']'
   + kill -s TERM 62800
   ++ jps -vl
   ++ grep java
   ++ grep -i QuorumPeerMain
   ++ grep -v grep
   ++ awk '{print $1}'
   + PIDS=
   + '[' '!' -z '' ']'
   + mv -f 
${FLINK_DIR}/flink-dist/target/flink-1.7-SNAPSHOT-bin/flink-1.7-SNAPSHOT/conf/flink-conf.yaml.bak
 
${FLINK_DIR}/flink-dist/target/flink-1.7-SNAPSHOT-bin/flink-1.7-SNAPSHOT/conf/flink-conf.yaml
   
   [PASS] 'Modern Kafka end-to-end test' passed after 1 minutes and 0 seconds! 
Test exited with exit code 0.
   
   Stopping taskexecutor daemon (pid: 64351) on host ${MY_HOST}.
   Stopping standalonesession daemon (pid: 63917) on host ${MY_HOST}.
   No zookeeper daemon to stop on host ${MY_HOST}.
   Deleted all files under 
${FLINK_DIR}/flink-dist/target/flink-1.7-SNAPSHOT-bin/flink-1.7-SNAPSHOT/log/
   Deleted 
${FLINK_DIR}/flink-end-to-end-tests/test-scripts/temp-test-directory-21N
   
   [PASS] All tests passed
   ```
   
   the problem (see comment in this function): 
   
   ```shell
   
   function stop_kafka_cluster {
 $KAFKA_DIR/bin/kafka-server-stop.sh
 $KAFKA_DIR/bin/zookeeper-server-stop.sh
   
 PIDS=$(jps -vl | grep -i 'kafka\.Kafka' | grep java | grep -v grep | awk 
'{print $1}')
   
 if [ ! -z "$PIDS" ]; then
   kill -s TERM $PIDS
 fi
   
 # here
 PIDS=$(jps -vl | grep java | grep -i QuorumPeerMain | grep -v grep | awk 
'{print $1}') 
   
 if [ ! -z "$PIDS" ]; then
   kill -s TERM $PIDS
 fi
   }
   ```


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


> Provide End-to-end test cases for modern Kafka connectors
> -
>
> Key: FLINK-10600
> URL: https://issues.apache.org/jira/browse/FLINK-10600
> Project: Flink
>  Issue Type: Sub-task
>  Components: Kafka Connector
>Reporter: vinoyang
>Assignee: vinoyang
>Priority: Blocker
>  Labels: pull-request-available
> Fix For: 1.7.0
>
>




--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[GitHub] yanghua edited a comment on issue #6924: [FLINK-10600] Provide End-to-end test cases for modern Kafka connectors

[GitBox]

yanghua edited a comment on issue #6924: [FLINK-10600] Provide End-to-end test 
cases for modern Kafka connectors
URL: https://github.com/apache/flink/pull/6924#issuecomment-434539461
 
 
   @pnowojski I can give you more details : 
   
   added `set -x` and `set -e`: 
   
   ```
   + stop_kafka_cluster
   + 
${FLINK_DIR}/flink-end-to-end-tests/test-scripts/temp-test-directory-01N/kafka_2.11-2.0.0/bin/kafka-server-stop.sh
   + 
${FLINK_DIR}/flink-end-to-end-tests/test-scripts/temp-test-directory-01N/kafka_2.11-2.0.0/bin/zookeeper-server-stop.sh
   ++ grep -i 'kafka\.Kafka'
   ++ jps -vl
   ++ grep java
   ++ grep -v grep
   ++ awk '{print $1}'
   + PIDS=55405
   + '[' '!' -z 55405 ']'
   + kill -s TERM 55405
   ++ jps -vl
   ++ grep java
   ++ grep -i QuorumPeerMain
   ++ grep -v grep
   ++ awk '{print $1}'
   + PIDS=
   
   [FAIL] 'Modern Kafka end-to-end test' failed after 0 minutes and 57 seconds! 
Test exited with exit code 1
   
   Stopping taskexecutor daemon (pid: 56949) on host ${MY_HOST}.
   Stopping standalonesession daemon (pid: 56518) on host ${MY_HOST}.
   No zookeeper daemon to stop on host ${MY_HOST}.
   ```
   
   But, just add `set -x`
   
   ```
   + stop_kafka_cluster
   + 
${FLINK_DIR}/flink-end-to-end-tests/test-scripts/temp-test-directory-21N/kafka_2.11-2.0.0/bin/kafka-server-stop.sh
   + 
${FLINK_DIR}/flink-end-to-end-tests/test-scripts/temp-test-directory-21N/kafka_2.11-2.0.0/bin/zookeeper-server-stop.sh
   ++ jps -vl
   ++ grep -i 'kafka\.Kafka'
   ++ grep java
   ++ grep -v grep
   ++ awk '{print $1}'
   + PIDS=62800
   + '[' '!' -z 62800 ']'
   + kill -s TERM 62800
   ++ jps -vl
   ++ grep java
   ++ grep -i QuorumPeerMain
   ++ grep -v grep
   ++ awk '{print $1}'
   + PIDS=
   + '[' '!' -z '' ']'
   + mv -f 
${FLINK_DIR}/flink-dist/target/flink-1.7-SNAPSHOT-bin/flink-1.7-SNAPSHOT/conf/flink-conf.yaml.bak
 
${FLINK_DIR}/flink-dist/target/flink-1.7-SNAPSHOT-bin/flink-1.7-SNAPSHOT/conf/flink-conf.yaml
   
   [PASS] 'Modern Kafka end-to-end test' passed after 1 minutes and 0 seconds! 
Test exited with exit code 0.
   
   Stopping taskexecutor daemon (pid: 64351) on host ${MY_HOST}.
   Stopping standalonesession daemon (pid: 63917) on host ${MY_HOST}.
   No zookeeper daemon to stop on host ${MY_HOST}.
   Deleted all files under 
${FLINK_DIR}/flink-dist/target/flink-1.7-SNAPSHOT-bin/flink-1.7-SNAPSHOT/log/
   Deleted 
${FLINK_DIR}/flink-end-to-end-tests/test-scripts/temp-test-directory-21N
   
   [PASS] All tests passed
   ```
   
   the problem (see comment in this function): 
   
   ```shell
   
   function stop_kafka_cluster {
 $KAFKA_DIR/bin/kafka-server-stop.sh
 $KAFKA_DIR/bin/zookeeper-server-stop.sh
   
 PIDS=$(jps -vl | grep -i 'kafka\.Kafka' | grep java | grep -v grep | awk 
'{print $1}')
   
 if [ ! -z "$PIDS" ]; then
   kill -s TERM $PIDS
 fi
   
 # here
 PIDS=$(jps -vl | grep java | grep -i QuorumPeerMain | grep -v grep | awk 
'{print $1}') 
   
 if [ ! -z "$PIDS" ]; then
   kill -s TERM $PIDS
 fi
   }
   ```


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-10600) Provide End-to-end test cases for modern Kafka connectors

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10600:


yanghua commented on issue #6924: [FLINK-10600] Provide End-to-end test cases 
for modern Kafka connectors
URL: https://github.com/apache/flink/pull/6924#issuecomment-434539461
 
 
   @pnowojski I can give you more details : 
   
   added `set -x` and `set -e`: 
   
   ```
   + stop_kafka_cluster
   + 
${FLINK_DIR}/flink-end-to-end-tests/test-scripts/temp-test-directory-01N/kafka_2.11-2.0.0/bin/kafka-server-stop.sh
   + 
${FLINK_DIR}/flink-end-to-end-tests/test-scripts/temp-test-directory-01N/kafka_2.11-2.0.0/bin/zookeeper-server-stop.sh
   ++ grep -i 'kafka\.Kafka'
   ++ jps -vl
   ++ grep java
   ++ grep -v grep
   ++ awk '{print $1}'
   + PIDS=55405
   + '[' '!' -z 55405 ']'
   + kill -s TERM 55405
   ++ jps -vl
   ++ grep java
   ++ grep -i QuorumPeerMain
   ++ grep -v grep
   ++ awk '{print $1}'
   + PIDS=
   
   [FAIL] 'Modern Kafka end-to-end test' failed after 0 minutes and 57 seconds! 
Test exited with exit code 1
   
   Stopping taskexecutor daemon (pid: 56949) on host ${MY_HOST}.
   Stopping standalonesession daemon (pid: 56518) on host ${MY_HOST}.
   No zookeeper daemon to stop on host ${MY_HOST}.
   ```
   
   But, just add `set -x`
   
   ```
   + stop_kafka_cluster
   + 
${FLINK_DIR}/flink-end-to-end-tests/test-scripts/temp-test-directory-21N/kafka_2.11-2.0.0/bin/kafka-server-stop.sh
   + 
${FLINK_DIR}/flink-end-to-end-tests/test-scripts/temp-test-directory-21N/kafka_2.11-2.0.0/bin/zookeeper-server-stop.sh
   ++ jps -vl
   ++ grep -i 'kafka\.Kafka'
   ++ grep java
   ++ grep -v grep
   ++ awk '{print $1}'
   + PIDS=62800
   + '[' '!' -z 62800 ']'
   + kill -s TERM 62800
   ++ jps -vl
   ++ grep java
   ++ grep -i QuorumPeerMain
   ++ grep -v grep
   ++ awk '{print $1}'
   + PIDS=
   + '[' '!' -z '' ']'
   + mv -f 
${FLINK_DIR}/flink-dist/target/flink-1.7-SNAPSHOT-bin/flink-1.7-SNAPSHOT/conf/flink-conf.yaml.bak
 
${FLINK_DIR}/flink-dist/target/flink-1.7-SNAPSHOT-bin/flink-1.7-SNAPSHOT/conf/flink-conf.yaml
   
   [PASS] 'Modern Kafka end-to-end test' passed after 1 minutes and 0 seconds! 
Test exited with exit code 0.
   
   Stopping taskexecutor daemon (pid: 64351) on host ${MY_HOST}.
   Stopping standalonesession daemon (pid: 63917) on host ${MY_HOST}.
   No zookeeper daemon to stop on host ${MY_HOST}.
   Deleted all files under 
${FLINK_DIR}/flink-dist/target/flink-1.7-SNAPSHOT-bin/flink-1.7-SNAPSHOT/log/
   Deleted 
${FLINK_DIR}/flink-end-to-end-tests/test-scripts/temp-test-directory-21N
   
   [PASS] All tests passed
   ```
   
   the problem : 
   
   ```shell
   
   function stop_kafka_cluster {
 $KAFKA_DIR/bin/kafka-server-stop.sh
 $KAFKA_DIR/bin/zookeeper-server-stop.sh
   
 PIDS=$(jps -vl | grep -i 'kafka\.Kafka' | grep java | grep -v grep | awk 
'{print $1}')
   
 if [ ! -z "$PIDS" ]; then
   kill -s TERM $PIDS
 fi
   
 # here
 PIDS=$(jps -vl | grep java | grep -i QuorumPeerMain | grep -v grep | awk 
'{print $1}') 
   
 if [ ! -z "$PIDS" ]; then
   kill -s TERM $PIDS
 fi
   }
   ```


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


> Provide End-to-end test cases for modern Kafka connectors
> -
>
> Key: FLINK-10600
> URL: https://issues.apache.org/jira/browse/FLINK-10600
> Project: Flink
>  Issue Type: Sub-task
>  Components: Kafka Connector
>Reporter: vinoyang
>Assignee: vinoyang
>Priority: Blocker
>  Labels: pull-request-available
> Fix For: 1.7.0
>
>




--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[GitHub] yanghua commented on issue #6924: [FLINK-10600] Provide End-to-end test cases for modern Kafka connectors

[GitBox]

yanghua commented on issue #6924: [FLINK-10600] Provide End-to-end test cases 
for modern Kafka connectors
URL: https://github.com/apache/flink/pull/6924#issuecomment-434539461
 
 
   @pnowojski I can give you more details : 
   
   added `set -x` and `set -e`: 
   
   ```
   + stop_kafka_cluster
   + 
${FLINK_DIR}/flink-end-to-end-tests/test-scripts/temp-test-directory-01N/kafka_2.11-2.0.0/bin/kafka-server-stop.sh
   + 
${FLINK_DIR}/flink-end-to-end-tests/test-scripts/temp-test-directory-01N/kafka_2.11-2.0.0/bin/zookeeper-server-stop.sh
   ++ grep -i 'kafka\.Kafka'
   ++ jps -vl
   ++ grep java
   ++ grep -v grep
   ++ awk '{print $1}'
   + PIDS=55405
   + '[' '!' -z 55405 ']'
   + kill -s TERM 55405
   ++ jps -vl
   ++ grep java
   ++ grep -i QuorumPeerMain
   ++ grep -v grep
   ++ awk '{print $1}'
   + PIDS=
   
   [FAIL] 'Modern Kafka end-to-end test' failed after 0 minutes and 57 seconds! 
Test exited with exit code 1
   
   Stopping taskexecutor daemon (pid: 56949) on host ${MY_HOST}.
   Stopping standalonesession daemon (pid: 56518) on host ${MY_HOST}.
   No zookeeper daemon to stop on host ${MY_HOST}.
   ```
   
   But, just add `set -x`
   
   ```
   + stop_kafka_cluster
   + 
${FLINK_DIR}/flink-end-to-end-tests/test-scripts/temp-test-directory-21N/kafka_2.11-2.0.0/bin/kafka-server-stop.sh
   + 
${FLINK_DIR}/flink-end-to-end-tests/test-scripts/temp-test-directory-21N/kafka_2.11-2.0.0/bin/zookeeper-server-stop.sh
   ++ jps -vl
   ++ grep -i 'kafka\.Kafka'
   ++ grep java
   ++ grep -v grep
   ++ awk '{print $1}'
   + PIDS=62800
   + '[' '!' -z 62800 ']'
   + kill -s TERM 62800
   ++ jps -vl
   ++ grep java
   ++ grep -i QuorumPeerMain
   ++ grep -v grep
   ++ awk '{print $1}'
   + PIDS=
   + '[' '!' -z '' ']'
   + mv -f 
${FLINK_DIR}/flink-dist/target/flink-1.7-SNAPSHOT-bin/flink-1.7-SNAPSHOT/conf/flink-conf.yaml.bak
 
${FLINK_DIR}/flink-dist/target/flink-1.7-SNAPSHOT-bin/flink-1.7-SNAPSHOT/conf/flink-conf.yaml
   
   [PASS] 'Modern Kafka end-to-end test' passed after 1 minutes and 0 seconds! 
Test exited with exit code 0.
   
   Stopping taskexecutor daemon (pid: 64351) on host ${MY_HOST}.
   Stopping standalonesession daemon (pid: 63917) on host ${MY_HOST}.
   No zookeeper daemon to stop on host ${MY_HOST}.
   Deleted all files under 
${FLINK_DIR}/flink-dist/target/flink-1.7-SNAPSHOT-bin/flink-1.7-SNAPSHOT/log/
   Deleted 
${FLINK_DIR}/flink-end-to-end-tests/test-scripts/temp-test-directory-21N
   
   [PASS] All tests passed
   ```
   
   the problem : 
   
   ```shell
   
   function stop_kafka_cluster {
 $KAFKA_DIR/bin/kafka-server-stop.sh
 $KAFKA_DIR/bin/zookeeper-server-stop.sh
   
 PIDS=$(jps -vl | grep -i 'kafka\.Kafka' | grep java | grep -v grep | awk 
'{print $1}')
   
 if [ ! -z "$PIDS" ]; then
   kill -s TERM $PIDS
 fi
   
 # here
 PIDS=$(jps -vl | grep java | grep -i QuorumPeerMain | grep -v grep | awk 
'{print $1}') 
   
 if [ ! -z "$PIDS" ]; then
   kill -s TERM $PIDS
 fi
   }
   ```


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-2028) Download pages must not link to snapshots

[Sebb (JIRA)]

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

Sebb commented on FLINK-2028:
-

SNAPSHOT builds should only be advertised to Flink developers, so should only 
be mentioned on pages intended for them

> Download pages must not link to snapshots
> -
>
> Key: FLINK-2028
> URL: https://issues.apache.org/jira/browse/FLINK-2028
> Project: Flink
>  Issue Type: Bug
>Reporter: Sebb
>Assignee: Ufuk Celebi
>Priority: Major
>  Labels: pull-request-available
>
> Snapshot builds that have not passed a PMC vote must not be advertised to the 
> general public, so must not be included on the downloads page.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-10672) Task stuck while writing output to flink

[Ankur Goenka (JIRA)]

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

Ankur Goenka commented on FLINK-10672:
--

Investigation notes:
 * Pipeline seems to be waiting on writing data. I verified back pressure in an 
stuck task. The consuming task was waiting on data while producing task was 
waiting to memorySegment to write the data.
 * Many of the broadcasting tasks were in created state and were not scheduled.
 * Using only BATCH execution mode did not help.
 * Using BATCH_FORCED execution mode makes the pipeline work. This suggests 
that we might not be identifying a breakPipeline correctly.
 * Pipeline seems to also work if we increase the taskManager memory. This 
might be because with higher memory, pipelined buffers can push all the output 
to memory before starting the consuming task.

 

> Task stuck while writing output to flink
> 
>
> Key: FLINK-10672
> URL: https://issues.apache.org/jira/browse/FLINK-10672
> Project: Flink
>  Issue Type: Bug
>Affects Versions: 1.5.4
> Environment: OS: Debuan rodente 4.17
> Flink version: 1.5.4
> ||Key||Value||
> |jobmanager.heap.mb|1024|
> |jobmanager.rpc.address|localhost|
> |jobmanager.rpc.port|6123|
> |metrics.reporter.jmx.class|org.apache.flink.metrics.jmx.JMXReporter|
> |metrics.reporter.jmx.port|9250-9260|
> |metrics.reporters|jmx|
> |parallelism.default|1|
> |rest.port|8081|
> |taskmanager.heap.mb|1024|
> |taskmanager.numberOfTaskSlots|1|
> |web.tmpdir|/tmp/flink-web-bdb73d6c-5b9e-47b5-9ebf-eed0a7c82c26|
>  
> h1. Overview
> ||Data Port||All Slots||Free Slots||CPU Cores||Physical Memory||JVM Heap 
> Size||Flink Managed Memory||
> |43501|1|0|12|62.9 GB|922 MB|642 MB|
> h1. Memory
> h2. JVM (Heap/Non-Heap)
> ||Type||Committed||Used||Maximum||
> |Heap|922 MB|575 MB|922 MB|
> |Non-Heap|68.8 MB|64.3 MB|-1 B|
> |Total|991 MB|639 MB|922 MB|
> h2. Outside JVM
> ||Type||Count||Used||Capacity||
> |Direct|3,292|105 MB|105 MB|
> |Mapped|0|0 B|0 B|
> h1. Network
> h2. Memory Segments
> ||Type||Count||
> |Available|3,194|
> |Total|3,278|
> h1. Garbage Collection
> ||Collector||Count||Time||
> |G1_Young_Generation|13|336|
> |G1_Old_Generation|1|21|
>Reporter: Ankur Goenka
>Priority: Major
> Attachments: 1uruvakHxBu.png, 3aDKQ24WvKk.png, Po89UGDn58V.png, 
> jmx_dump.json, jmx_dump_detailed.json, jstack_129827.log, jstack_163822.log, 
> jstack_66985.log
>
>
> I am running a fairly complex pipleline with 200+ task.
> The pipeline works fine with small data (order of 10kb input) but gets stuck 
> with a slightly larger data (300kb input).
>  
> The task gets stuck while writing the output toFlink, more specifically it 
> gets stuck while requesting memory segment in local buffer pool. The Task 
> manager UI shows that it has enough memory and memory segments to work with.
> The relevant stack trace is 
> {quote}"grpc-default-executor-0" #138 daemon prio=5 os_prio=0 
> tid=0x7fedb0163800 nid=0x30b7f in Object.wait() [0x7fedb4f9]
>  java.lang.Thread.State: TIMED_WAITING (on object monitor)
>  at (C/C++) 0x7fef201c7dae (Unknown Source)
>  at (C/C++) 0x7fef1f2aea07 (Unknown Source)
>  at (C/C++) 0x7fef1f241cd3 (Unknown Source)
>  at java.lang.Object.wait(Native Method)
>  - waiting on <0xf6d56450> (a java.util.ArrayDeque)
>  at 
> org.apache.flink.runtime.io.network.buffer.LocalBufferPool.requestMemorySegment(LocalBufferPool.java:247)
>  - locked <0xf6d56450> (a java.util.ArrayDeque)
>  at 
> org.apache.flink.runtime.io.network.buffer.LocalBufferPool.requestBufferBuilderBlocking(LocalBufferPool.java:204)
>  at 
> org.apache.flink.runtime.io.network.api.writer.RecordWriter.requestNewBufferBuilder(RecordWriter.java:213)
>  at 
> org.apache.flink.runtime.io.network.api.writer.RecordWriter.sendToTarget(RecordWriter.java:144)
>  at 
> org.apache.flink.runtime.io.network.api.writer.RecordWriter.emit(RecordWriter.java:107)
>  at 
> org.apache.flink.runtime.operators.shipping.OutputCollector.collect(OutputCollector.java:65)
>  at 
> org.apache.flink.runtime.operators.util.metrics.CountingCollector.collect(CountingCollector.java:35)
>  at 
> org.apache.beam.runners.flink.translation.functions.FlinkExecutableStagePruningFunction.flatMap(FlinkExecutableStagePruningFunction.java:42)
>  at 
> org.apache.beam.runners.flink.translation.functions.FlinkExecutableStagePruningFunction.flatMap(FlinkExecutableStagePruningFunction.java:26)
>  at 
> org.apache.flink.runtime.operators.chaining.ChainedFlatMapDriver.collect(ChainedFlatMapDriver.java:80)
>  at 
> org.apache.flink.runtime.operators.util.metrics.CountingCollector.collect(CountingCollector.java:35)
>  at 
> 

[jira] [Commented] (FLINK-10728) Unexpected items on ASF mirrors

[Sebb (JIRA)]

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

Sebb commented on FLINK-10728:
--

I don't understand what you mean by "Flink itself is the only intended user".

Note that the 3rd party mirrors give their services for free. It's unfair to 
take up their space with code that is not intended for general public 
consumption. If the release is only intended for developers, then it should 
probably be stored elsewhere; please ask Infra.

> Unexpected items on ASF mirrors
> ---
>
> Key: FLINK-10728
> URL: https://issues.apache.org/jira/browse/FLINK-10728
> Project: Flink
>  Issue Type: Bug
> Environment: https://www.apache.org/dist/flink/flink-shaded-4.0/
> https://www.apache.org/dist/flink/flink-shaded-5.0/
>Reporter: Sebb
>Assignee: Chesnay Schepler
>Priority: Major
>
> The flink-shaded artifacts are not linked from the download page, and don't 
> appear to have been announced on any ASF mailing lists, so it's not clear why 
> they are on the ASF mirrors.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Updated] (FLINK-9953) Active Kubernetes integration

[JIN SUN (JIRA)]

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

JIN SUN updated FLINK-9953:
---
Description: 
This is the umbrella issue tracking Flink's active Kubernetes integration. 
Active means in this context that the {{ResourceManager}} can talk to 
Kubernetes to launch new pods similar to Flink's Yarn and Mesos integration.

 

Document can be found here: 
[https://docs.google.com/document/d/1Zmhui_29VASPcBOEqyMWnF3L6WEWZ4kedrCqya0WaAk/edit?usp=sharing]
 

  was:This is the umbrella issue tracking Flink's active Kubernetes 
integration. Active means in this context that the {{ResourceManager}} can talk 
to Kubernetes to launch new pods similar to Flink's Yarn and Mesos integration.


> Active Kubernetes integration
> -
>
> Key: FLINK-9953
> URL: https://issues.apache.org/jira/browse/FLINK-9953
> Project: Flink
>  Issue Type: New Feature
>  Components: Distributed Coordination, ResourceManager
>Reporter: Till Rohrmann
>Assignee: JIN SUN
>Priority: Major
> Fix For: 1.7.0
>
>
> This is the umbrella issue tracking Flink's active Kubernetes integration. 
> Active means in this context that the {{ResourceManager}} can talk to 
> Kubernetes to launch new pods similar to Flink's Yarn and Mesos integration.
>  
> Document can be found here: 
> [https://docs.google.com/document/d/1Zmhui_29VASPcBOEqyMWnF3L6WEWZ4kedrCqya0WaAk/edit?usp=sharing]
>  



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (FLINK-10729) Create a Hive a connector to access Hive data

[Xuefu Zhang (JIRA)]

Xuefu Zhang created FLINK-10729:
---

 Summary: Create a Hive a connector to access Hive data
 Key: FLINK-10729
 URL: https://issues.apache.org/jira/browse/FLINK-10729
 Project: Flink
  Issue Type: Sub-task
  Components: Table API  SQL
Affects Versions: 1.6.2
Reporter: Xuefu Zhang


As part of Flink-Hive integration effort, it's important for Flink to access 
(read/write) Hive data, which is the responsibility of Hive connector. While 
there is a HCatalog data connector in the code base, it's not complete (i.e. 
missing all connector related classes such as validators, etc.). Further, 
HCatalog interface has many limitations such as accessing a subset of Hive 
data, supporting a subset of Hive data types, etc. In addition, it's not 
actively maintained. In fact, it's now only a sub-project in Hive.

Therefore, here we propose a complete connector set for Hive tables, not via 
HCatalog, but via direct Hive interface. HCatalog connector will be deprecated.

Please note that connector on Hive metadata is already covered in other JIRAs, 
as {{HiveExternalCatalog}}.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Updated] (FLINK-10556) Integration with Apache Hive

[Xuefu Zhang (JIRA)]

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

Xuefu Zhang updated FLINK-10556:

Attachment: (was: Flink-Hive Metastore Connectivity Design.pdf)

> Integration with Apache Hive
> 
>
> Key: FLINK-10556
> URL: https://issues.apache.org/jira/browse/FLINK-10556
> Project: Flink
>  Issue Type: New Feature
>  Components: Batch Connectors and Input/Output Formats, SQL Client, 
> Table API  SQL
>Affects Versions: 1.6.0
>Reporter: Xuefu Zhang
>Assignee: Xuefu Zhang
>Priority: Major
> Attachments: Flink-Hive Metastore Connectivity Design.pdf, Proposal_ 
> Integrate Flink with Hive Ecosystem.pdf
>
>
> This is an umbrella JIRA tracking all enhancement and issues related to 
> integrating Flink with Hive ecosystem. This is an outcome of a discussion in 
> the community, and thanks go to everyone that provided feedback and interest.
> Specifically, we'd like to see the following features and capabilities 
> immediately in Flink:
> # Metadata interoperability
> # Data interoperability
> # Data type compatibility
> # Hive UDF support
> # DDL/DML/Query language compatibility
> For a longer term, we'd also like to add or improve:
> # Compatible SQL service, client tools, JDBC/ODBC drivers
> # Better task failure tolerance and task scheduling
> # Support other user customizations in Hive (storage handlers, serdes, etc).
> I will provide more details regarding the proposal in a doc shortly. Design 
> doc, if deemed necessary, will be provided in each related sub tasks under 
> this JIRA.
> Feedback and contributions are greatly welcome!



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Updated] (FLINK-10556) Integration with Apache Hive

[Xuefu Zhang (JIRA)]

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

Xuefu Zhang updated FLINK-10556:

Attachment: Flink-Hive Metastore Connectivity Design.pdf

> Integration with Apache Hive
> 
>
> Key: FLINK-10556
> URL: https://issues.apache.org/jira/browse/FLINK-10556
> Project: Flink
>  Issue Type: New Feature
>  Components: Batch Connectors and Input/Output Formats, SQL Client, 
> Table API  SQL
>Affects Versions: 1.6.0
>Reporter: Xuefu Zhang
>Assignee: Xuefu Zhang
>Priority: Major
> Attachments: Flink-Hive Metastore Connectivity Design.pdf, Proposal_ 
> Integrate Flink with Hive Ecosystem.pdf
>
>
> This is an umbrella JIRA tracking all enhancement and issues related to 
> integrating Flink with Hive ecosystem. This is an outcome of a discussion in 
> the community, and thanks go to everyone that provided feedback and interest.
> Specifically, we'd like to see the following features and capabilities 
> immediately in Flink:
> # Metadata interoperability
> # Data interoperability
> # Data type compatibility
> # Hive UDF support
> # DDL/DML/Query language compatibility
> For a longer term, we'd also like to add or improve:
> # Compatible SQL service, client tools, JDBC/ODBC drivers
> # Better task failure tolerance and task scheduling
> # Support other user customizations in Hive (storage handlers, serdes, etc).
> I will provide more details regarding the proposal in a doc shortly. Design 
> doc, if deemed necessary, will be provided in each related sub tasks under 
> this JIRA.
> Feedback and contributions are greatly welcome!



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Assigned] (FLINK-10699) Create a catalog implementation for persistent Flink meta objects using Hive metastore as a registry

[Bowen Li (JIRA)]

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

Bowen Li reassigned FLINK-10699:


Assignee: Bowen Li

> Create a catalog implementation for persistent Flink meta objects using Hive 
> metastore as a registry
> 
>
> Key: FLINK-10699
> URL: https://issues.apache.org/jira/browse/FLINK-10699
> Project: Flink
>  Issue Type: Sub-task
>  Components: Table API  SQL
>Affects Versions: 1.6.1
>Reporter: Xuefu Zhang
>Assignee: Bowen Li
>Priority: Major
>
> Similar to FLINK-10697, but using Hive metastore as persistent storage.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[GitHub] yxu-valleytider opened a new pull request #6968: Consuming data from DynamoDB streams to flink

[GitBox]

yxu-valleytider opened a new pull request #6968: Consuming data from DynamoDB 
streams to flink
URL: https://github.com/apache/flink/pull/6968
 
 
   ## What is the purpose of the change
   
   This PR introduces a new Flink source to consume directly from [dynamodb 
streams](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html).
 This new source is built on top of the existing Kinesis connector. It 
interacts with the dynamodb streams via a 
[dynamodb-streams-kinesis-adapter](https://github.com/awslabs/dynamodb-streams-kinesis-adapter)
 client.
   
   ## Brief change log
   
   New data stream can be constructed to directly pull data from DynamoDB 
streams. 
   
   DataStream dynamodbStreams = env.addSource(new 
FlinkDynamodbStreamsConsumer(streamName, ..., ...))
   
   Other changes include:
   
 - *DynamodbStreamsProxy class which uses a `dynamodbstreams-kinesis` 
adapter client to interact with Dynamodb streams.*
 - *Port the `describeStream` API from the flink-1.5 branch into 
DynamodbStreamsProxy.*
 - *Minimalist DynamodbStreamsSchema class which helps deserialize dynamodb 
streams records*
   
   
   ## Verifying this change
   
   This change is already covered by most of the existing Flink kinesis 
connector tests.
   
   Manual tests are provided to verify that the dynamodbstreams connector can: 
 - *consume from a stream given its ARN*
 - *consume from a given table with new stream created*
   
   ## Does this pull request potentially affect one of the following parts:
   
 - Dependencies (does it add or upgrade a dependency): (yes) 
Added optional dependency on AWS 
[dynamodb-streams-kinesis-adapter](https://github.com/awslabs/dynamodb-streams-kinesis-adapter).
 
 - The public API, i.e., is any changed class annotated with 
`@Public(Evolving)`: (no)
 - The serializers: (no)
 - The runtime per-record code paths (performance sensitive): (no)
 - Anything that affects deployment or recovery: JobManager (and its 
components), Checkpointing, Yarn/Mesos, ZooKeeper: (no)
 - The S3 file system connector: (no)
   
   ## Documentation
   
 - Does this pull request introduce a new feature? (yes)
 - If yes, how is the feature documented? (JavaDocs)
   


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-10166) Dependency problems when executing SQL query in sql-client

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10166:


igalshilman commented on a change in pull request #6966: [FLINK-10166] [table] 
Reduce dependencies by removing org.apache.commons
URL: https://github.com/apache/flink/pull/6966#discussion_r229473954
 
 

 ##
 File path: 
flink-libraries/flink-table-common/src/main/java/org/apache/flink/table/utils/EncodingUtils.java
 ##
 @@ -40,14 +44,18 @@
 
private static final Base64.Decoder BASE64_DECODER = 
java.util.Base64.getUrlDecoder();
 
+   private static final MessageDigest MD5_MESSAGE_DIGEST = 
getMd5MessageDigest();
 
 Review comment:
   Would MD5_MESSAGE_DIGEST be accessed concurrently? afaik this is not thread 
safe.


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


> Dependency problems when executing SQL query in sql-client
> --
>
> Key: FLINK-10166
> URL: https://issues.apache.org/jira/browse/FLINK-10166
> Project: Flink
>  Issue Type: Bug
>  Components: Table API  SQL
>Affects Versions: 1.6.0
>Reporter: Dawid Wysakowicz
>Assignee: Timo Walther
>Priority: Blocker
>  Labels: pull-request-available
> Fix For: 1.7.0
>
>
> When tried to run query:
> {code}
> select count(distinct name) from (Values ('a'), ('b')) AS NameTable(name)
> {code}
> in {{sql-client.sh}} I got:
> {code}
> [ERROR] Could not execute SQL statement. Reason:
> org.codehaus.commons.compiler.CompileException: Line 43, Column 10: Unknown 
> variable or type "org.apache.commons.codec.binary.Base64"
> {code}



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-10166) Dependency problems when executing SQL query in sql-client

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10166:


igalshilman commented on a change in pull request #6966: [FLINK-10166] [table] 
Reduce dependencies by removing org.apache.commons
URL: https://github.com/apache/flink/pull/6966#discussion_r229481767
 
 

 ##
 File path: 
flink-libraries/flink-table-common/src/main/java/org/apache/flink/table/utils/EncodingUtils.java
 ##
 @@ -87,10 +95,145 @@ public static String encodeObjectToString(Serializable 
obj) {
return loadClass(qualifiedName, 
Thread.currentThread().getContextClassLoader());
}
 
+   public static String encodeStringToBase64(String string) {
+   return new 
String(java.util.Base64.getEncoder().encode(string.getBytes(UTF_8)), UTF_8);
+   }
+
+   public static String decodeBase64ToString(String base64) {
+   return new 
String(java.util.Base64.getDecoder().decode(base64.getBytes(UTF_8)), UTF_8);
+   }
+
+   public static byte[] md5(String string) {
+   if (MD5_MESSAGE_DIGEST == null) {
+   throw new TableException("Unsupported MD5 algorithm.");
+   }
+   return MD5_MESSAGE_DIGEST.digest(string.getBytes(UTF_8));
+   }
+
+   public static String hex(String string) {
+   return hex(string.getBytes(UTF_8));
+   }
+
+   public static String hex(byte[] bytes) {
+   // adopted from https://stackoverflow.com/a/9855338
+   final char[] hexChars = new char[bytes.length * 2];
+   for (int j = 0; j < bytes.length; j++) {
+   final int v = bytes[j] & 0xFF;
+   hexChars[j * 2] = HEX_CHARS[v >>> 4];
+   hexChars[j * 2 + 1] = HEX_CHARS[v & 0x0F];
+   }
+   return new String(hexChars);
+   }
+
+   private static MessageDigest getMd5MessageDigest() {
+   try {
+   return MessageDigest.getInstance("MD5");
+   } catch (NoSuchAlgorithmException e) {
+   return null;
+   }
+   }
+
+   // 

+   // Java String Repetition
+   //
+   // copied from o.a.commons.lang3.StringUtils (commons-lang3:3.3.2)
+   // 

+
+   private static final String EMPTY = "";
+
+   /**
+* The maximum size to which the padding constant(s) can expand.
+*/
+   private static final int PAD_LIMIT = 8192;
+
+   /**
+* Repeat a String {@code repeat} times to form a new String.
+*
+* 
+* StringUtils.repeat(null, 2) = null
+* StringUtils.repeat("", 0)   = ""
+* StringUtils.repeat("", 2)   = ""
+* StringUtils.repeat("a", 3)  = "aaa"
+* StringUtils.repeat("ab", 2) = "abab"
+* StringUtils.repeat("a", -2) = ""
+* 
+*
+* @param strthe String to repeat, may be null
+* @param repeat number of times to repeat str, negative treated as zero
+* @return a new String consisting of the original String repeated, 
{@code null} if null String input
+*/
+   public static String repeat(final String str, final int repeat) {
+   // Performance tuned for 2.0 (JDK1.4)
+
+   if (str == null) {
+   return null;
+   }
+   if (repeat <= 0) {
+   return EMPTY;
+   }
+   final int inputLength = str.length();
+   if (repeat == 1 || inputLength == 0) {
+   return str;
+   }
+   if (inputLength == 1 && repeat <= PAD_LIMIT) {
+   return repeat(str.charAt(0), repeat);
+   }
+
+   final int outputLength = inputLength * repeat;
+   switch (inputLength) {
+   case 1:
+   return repeat(str.charAt(0), repeat);
+   case 2:
+   final char ch0 = str.charAt(0);
+   final char ch1 = str.charAt(1);
+   final char[] output2 = new char[outputLength];
+   for (int i = repeat * 2 - 2; i >= 0; i--, i--) {
+   output2[i] = ch0;
+   output2[i + 1] = ch1;
+   }
+   return new String(output2);
+   default:
+   final StringBuilder buf = new 
StringBuilder(outputLength);
+  

[jira] [Commented] (FLINK-10166) Dependency problems when executing SQL query in sql-client

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10166:


igalshilman commented on a change in pull request #6966: [FLINK-10166] [table] 
Reduce dependencies by removing org.apache.commons
URL: https://github.com/apache/flink/pull/6966#discussion_r229487810
 
 

 ##
 File path: 
flink-libraries/flink-table-common/src/main/java/org/apache/flink/table/utils/EncodingUtils.java
 ##
 @@ -87,10 +95,145 @@ public static String encodeObjectToString(Serializable 
obj) {
return loadClass(qualifiedName, 
Thread.currentThread().getContextClassLoader());
}
 
+   public static String encodeStringToBase64(String string) {
+   return new 
String(java.util.Base64.getEncoder().encode(string.getBytes(UTF_8)), UTF_8);
+   }
+
+   public static String decodeBase64ToString(String base64) {
+   return new 
String(java.util.Base64.getDecoder().decode(base64.getBytes(UTF_8)), UTF_8);
+   }
+
+   public static byte[] md5(String string) {
+   if (MD5_MESSAGE_DIGEST == null) {
+   throw new TableException("Unsupported MD5 algorithm.");
+   }
+   return MD5_MESSAGE_DIGEST.digest(string.getBytes(UTF_8));
+   }
+
+   public static String hex(String string) {
+   return hex(string.getBytes(UTF_8));
+   }
+
+   public static String hex(byte[] bytes) {
+   // adopted from https://stackoverflow.com/a/9855338
+   final char[] hexChars = new char[bytes.length * 2];
+   for (int j = 0; j < bytes.length; j++) {
+   final int v = bytes[j] & 0xFF;
+   hexChars[j * 2] = HEX_CHARS[v >>> 4];
+   hexChars[j * 2 + 1] = HEX_CHARS[v & 0x0F];
+   }
+   return new String(hexChars);
+   }
+
+   private static MessageDigest getMd5MessageDigest() {
+   try {
+   return MessageDigest.getInstance("MD5");
+   } catch (NoSuchAlgorithmException e) {
+   return null;
+   }
+   }
+
+   // 

+   // Java String Repetition
+   //
+   // copied from o.a.commons.lang3.StringUtils (commons-lang3:3.3.2)
+   // 

+
+   private static final String EMPTY = "";
+
+   /**
+* The maximum size to which the padding constant(s) can expand.
+*/
+   private static final int PAD_LIMIT = 8192;
+
+   /**
+* Repeat a String {@code repeat} times to form a new String.
+*
+* 
+* StringUtils.repeat(null, 2) = null
+* StringUtils.repeat("", 0)   = ""
+* StringUtils.repeat("", 2)   = ""
+* StringUtils.repeat("a", 3)  = "aaa"
+* StringUtils.repeat("ab", 2) = "abab"
+* StringUtils.repeat("a", -2) = ""
+* 
+*
+* @param strthe String to repeat, may be null
+* @param repeat number of times to repeat str, negative treated as zero
+* @return a new String consisting of the original String repeated, 
{@code null} if null String input
+*/
+   public static String repeat(final String str, final int repeat) {
+   // Performance tuned for 2.0 (JDK1.4)
+
+   if (str == null) {
+   return null;
+   }
+   if (repeat <= 0) {
+   return EMPTY;
+   }
+   final int inputLength = str.length();
+   if (repeat == 1 || inputLength == 0) {
+   return str;
+   }
+   if (inputLength == 1 && repeat <= PAD_LIMIT) {
+   return repeat(str.charAt(0), repeat);
+   }
+
+   final int outputLength = inputLength * repeat;
+   switch (inputLength) {
+   case 1:
+   return repeat(str.charAt(0), repeat);
+   case 2:
+   final char ch0 = str.charAt(0);
+   final char ch1 = str.charAt(1);
+   final char[] output2 = new char[outputLength];
+   for (int i = repeat * 2 - 2; i >= 0; i--, i--) {
+   output2[i] = ch0;
+   output2[i + 1] = ch1;
+   }
+   return new String(output2);
+   default:
+   final StringBuilder buf = new 
StringBuilder(outputLength);
 
 

[jira] [Commented] (FLINK-9752) Add an S3 RecoverableWriter

[ASF GitHub Bot (JIRA)]

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

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

addisonj edited a comment on issue #6795: [FLINK-9752][s3-fs-connector] Add s3 
recoverable writer.
URL: https://github.com/apache/flink/pull/6795#issuecomment-434477664
 
 
   @kl0u the jira issue (https://issues.apache.org/jira/browse/FLINK-9752) says 
this is fixed in 1.6.2, but it doesn't appear like this has been backported to 
that release... Is this work somehow still reflected in 1.6.2?
   
   Edit: had copy pasted the wrong issue :)


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 an S3 RecoverableWriter
> ---
>
> Key: FLINK-9752
> URL: https://issues.apache.org/jira/browse/FLINK-9752
> Project: Flink
>  Issue Type: Sub-task
>  Components: Streaming Connectors
>Reporter: Stephan Ewen
>Assignee: Kostas Kloudas
>Priority: Major
>  Labels: pull-request-available
> Fix For: 1.6.2, 1.7.0
>
>
> S3 offers persistence only when uploads are complete. That means at the end 
> of simple uploads and uploads of parts of a MultiPartUpload.
> We should implement a RecoverableWriter for S3 that does a MultiPartUpload 
> with a Part per checkpoint.
> Recovering the reader needs the MultiPartUploadID and the list of ETags of 
> previous parts.
> We need additional staging of data in Flink state to work around the fact that
>  - Parts in a MultiPartUpload must be at least 5MB
>  - Part sizes must be known up front. (Note that data can still be streamed 
> in the upload)



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[GitHub] igalshilman commented on a change in pull request #6966: [FLINK-10166] [table] Reduce dependencies by removing org.apache.commons

[GitBox]

igalshilman commented on a change in pull request #6966: [FLINK-10166] [table] 
Reduce dependencies by removing org.apache.commons
URL: https://github.com/apache/flink/pull/6966#discussion_r229481767
 
 

 ##
 File path: 
flink-libraries/flink-table-common/src/main/java/org/apache/flink/table/utils/EncodingUtils.java
 ##
 @@ -87,10 +95,145 @@ public static String encodeObjectToString(Serializable 
obj) {
return loadClass(qualifiedName, 
Thread.currentThread().getContextClassLoader());
}
 
+   public static String encodeStringToBase64(String string) {
+   return new 
String(java.util.Base64.getEncoder().encode(string.getBytes(UTF_8)), UTF_8);
+   }
+
+   public static String decodeBase64ToString(String base64) {
+   return new 
String(java.util.Base64.getDecoder().decode(base64.getBytes(UTF_8)), UTF_8);
+   }
+
+   public static byte[] md5(String string) {
+   if (MD5_MESSAGE_DIGEST == null) {
+   throw new TableException("Unsupported MD5 algorithm.");
+   }
+   return MD5_MESSAGE_DIGEST.digest(string.getBytes(UTF_8));
+   }
+
+   public static String hex(String string) {
+   return hex(string.getBytes(UTF_8));
+   }
+
+   public static String hex(byte[] bytes) {
+   // adopted from https://stackoverflow.com/a/9855338
+   final char[] hexChars = new char[bytes.length * 2];
+   for (int j = 0; j < bytes.length; j++) {
+   final int v = bytes[j] & 0xFF;
+   hexChars[j * 2] = HEX_CHARS[v >>> 4];
+   hexChars[j * 2 + 1] = HEX_CHARS[v & 0x0F];
+   }
+   return new String(hexChars);
+   }
+
+   private static MessageDigest getMd5MessageDigest() {
+   try {
+   return MessageDigest.getInstance("MD5");
+   } catch (NoSuchAlgorithmException e) {
+   return null;
+   }
+   }
+
+   // 

+   // Java String Repetition
+   //
+   // copied from o.a.commons.lang3.StringUtils (commons-lang3:3.3.2)
+   // 

+
+   private static final String EMPTY = "";
+
+   /**
+* The maximum size to which the padding constant(s) can expand.
+*/
+   private static final int PAD_LIMIT = 8192;
+
+   /**
+* Repeat a String {@code repeat} times to form a new String.
+*
+* 
+* StringUtils.repeat(null, 2) = null
+* StringUtils.repeat("", 0)   = ""
+* StringUtils.repeat("", 2)   = ""
+* StringUtils.repeat("a", 3)  = "aaa"
+* StringUtils.repeat("ab", 2) = "abab"
+* StringUtils.repeat("a", -2) = ""
+* 
+*
+* @param strthe String to repeat, may be null
+* @param repeat number of times to repeat str, negative treated as zero
+* @return a new String consisting of the original String repeated, 
{@code null} if null String input
+*/
+   public static String repeat(final String str, final int repeat) {
+   // Performance tuned for 2.0 (JDK1.4)
+
+   if (str == null) {
+   return null;
+   }
+   if (repeat <= 0) {
+   return EMPTY;
+   }
+   final int inputLength = str.length();
+   if (repeat == 1 || inputLength == 0) {
+   return str;
+   }
+   if (inputLength == 1 && repeat <= PAD_LIMIT) {
+   return repeat(str.charAt(0), repeat);
+   }
+
+   final int outputLength = inputLength * repeat;
+   switch (inputLength) {
+   case 1:
+   return repeat(str.charAt(0), repeat);
+   case 2:
+   final char ch0 = str.charAt(0);
+   final char ch1 = str.charAt(1);
+   final char[] output2 = new char[outputLength];
+   for (int i = repeat * 2 - 2; i >= 0; i--, i--) {
+   output2[i] = ch0;
+   output2[i + 1] = ch1;
+   }
+   return new String(output2);
+   default:
+   final StringBuilder buf = new 
StringBuilder(outputLength);
+   for (int i = 0; i < repeat; i++) {
+   buf.append(str);
+   }
+   return buf.toString();
+   }
+   }
+
+  

[GitHub] addisonj edited a comment on issue #6795: [FLINK-9752][s3-fs-connector] Add s3 recoverable writer.

[GitBox]

addisonj edited a comment on issue #6795: [FLINK-9752][s3-fs-connector] Add s3 
recoverable writer.
URL: https://github.com/apache/flink/pull/6795#issuecomment-434477664
 
 
   @kl0u the jira issue (https://issues.apache.org/jira/browse/FLINK-9752) says 
this is fixed in 1.6.2, but it doesn't appear like this has been backported to 
that release... Is this work somehow still reflected in 1.6.2?
   
   Edit: had copy pasted the wrong issue :)


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] igalshilman commented on a change in pull request #6966: [FLINK-10166] [table] Reduce dependencies by removing org.apache.commons

[GitBox]

igalshilman commented on a change in pull request #6966: [FLINK-10166] [table] 
Reduce dependencies by removing org.apache.commons
URL: https://github.com/apache/flink/pull/6966#discussion_r229473954
 
 

 ##
 File path: 
flink-libraries/flink-table-common/src/main/java/org/apache/flink/table/utils/EncodingUtils.java
 ##
 @@ -40,14 +44,18 @@
 
private static final Base64.Decoder BASE64_DECODER = 
java.util.Base64.getUrlDecoder();
 
+   private static final MessageDigest MD5_MESSAGE_DIGEST = 
getMd5MessageDigest();
 
 Review comment:
   Would MD5_MESSAGE_DIGEST be accessed concurrently? afaik this is not thread 
safe.


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] igalshilman commented on a change in pull request #6966: [FLINK-10166] [table] Reduce dependencies by removing org.apache.commons

[GitBox]

igalshilman commented on a change in pull request #6966: [FLINK-10166] [table] 
Reduce dependencies by removing org.apache.commons
URL: https://github.com/apache/flink/pull/6966#discussion_r229487810
 
 

 ##
 File path: 
flink-libraries/flink-table-common/src/main/java/org/apache/flink/table/utils/EncodingUtils.java
 ##
 @@ -87,10 +95,145 @@ public static String encodeObjectToString(Serializable 
obj) {
return loadClass(qualifiedName, 
Thread.currentThread().getContextClassLoader());
}
 
+   public static String encodeStringToBase64(String string) {
+   return new 
String(java.util.Base64.getEncoder().encode(string.getBytes(UTF_8)), UTF_8);
+   }
+
+   public static String decodeBase64ToString(String base64) {
+   return new 
String(java.util.Base64.getDecoder().decode(base64.getBytes(UTF_8)), UTF_8);
+   }
+
+   public static byte[] md5(String string) {
+   if (MD5_MESSAGE_DIGEST == null) {
+   throw new TableException("Unsupported MD5 algorithm.");
+   }
+   return MD5_MESSAGE_DIGEST.digest(string.getBytes(UTF_8));
+   }
+
+   public static String hex(String string) {
+   return hex(string.getBytes(UTF_8));
+   }
+
+   public static String hex(byte[] bytes) {
+   // adopted from https://stackoverflow.com/a/9855338
+   final char[] hexChars = new char[bytes.length * 2];
+   for (int j = 0; j < bytes.length; j++) {
+   final int v = bytes[j] & 0xFF;
+   hexChars[j * 2] = HEX_CHARS[v >>> 4];
+   hexChars[j * 2 + 1] = HEX_CHARS[v & 0x0F];
+   }
+   return new String(hexChars);
+   }
+
+   private static MessageDigest getMd5MessageDigest() {
+   try {
+   return MessageDigest.getInstance("MD5");
+   } catch (NoSuchAlgorithmException e) {
+   return null;
+   }
+   }
+
+   // 

+   // Java String Repetition
+   //
+   // copied from o.a.commons.lang3.StringUtils (commons-lang3:3.3.2)
+   // 

+
+   private static final String EMPTY = "";
+
+   /**
+* The maximum size to which the padding constant(s) can expand.
+*/
+   private static final int PAD_LIMIT = 8192;
+
+   /**
+* Repeat a String {@code repeat} times to form a new String.
+*
+* 
+* StringUtils.repeat(null, 2) = null
+* StringUtils.repeat("", 0)   = ""
+* StringUtils.repeat("", 2)   = ""
+* StringUtils.repeat("a", 3)  = "aaa"
+* StringUtils.repeat("ab", 2) = "abab"
+* StringUtils.repeat("a", -2) = ""
+* 
+*
+* @param strthe String to repeat, may be null
+* @param repeat number of times to repeat str, negative treated as zero
+* @return a new String consisting of the original String repeated, 
{@code null} if null String input
+*/
+   public static String repeat(final String str, final int repeat) {
+   // Performance tuned for 2.0 (JDK1.4)
+
+   if (str == null) {
+   return null;
+   }
+   if (repeat <= 0) {
+   return EMPTY;
+   }
+   final int inputLength = str.length();
+   if (repeat == 1 || inputLength == 0) {
+   return str;
+   }
+   if (inputLength == 1 && repeat <= PAD_LIMIT) {
+   return repeat(str.charAt(0), repeat);
+   }
+
+   final int outputLength = inputLength * repeat;
+   switch (inputLength) {
+   case 1:
+   return repeat(str.charAt(0), repeat);
+   case 2:
+   final char ch0 = str.charAt(0);
+   final char ch1 = str.charAt(1);
+   final char[] output2 = new char[outputLength];
+   for (int i = repeat * 2 - 2; i >= 0; i--, i--) {
+   output2[i] = ch0;
+   output2[i + 1] = ch1;
+   }
+   return new String(output2);
+   default:
+   final StringBuilder buf = new 
StringBuilder(outputLength);
 
 Review comment:
   I know that you've just copied it, but honestly, I think that the default 
case is the only one that makes senes and all the complication in that method 
is a bit wired (as the comment above states - it is optimized for jdk1.4) 

[GitHub] addisonj commented on issue #6795: [FLINK-9752][s3-fs-connector] Add s3 recoverable writer.

[GitBox]

addisonj commented on issue #6795: [FLINK-9752][s3-fs-connector] Add s3 
recoverable writer.
URL: https://github.com/apache/flink/pull/6795#issuecomment-434477664
 
 
   @kl0u the jira issue (https://issues.apache.org/jira/browse/FLINK-10383) 
says this is fixed in 1.6.2, but it doesn't appear like this has been 
backported to that release... Is this work somehow still reflected in 1.6.2?


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-9752) Add an S3 RecoverableWriter

[ASF GitHub Bot (JIRA)]

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

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

addisonj commented on issue #6795: [FLINK-9752][s3-fs-connector] Add s3 
recoverable writer.
URL: https://github.com/apache/flink/pull/6795#issuecomment-434477664
 
 
   @kl0u the jira issue (https://issues.apache.org/jira/browse/FLINK-10383) 
says this is fixed in 1.6.2, but it doesn't appear like this has been 
backported to that release... Is this work somehow still reflected in 1.6.2?


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 an S3 RecoverableWriter
> ---
>
> Key: FLINK-9752
> URL: https://issues.apache.org/jira/browse/FLINK-9752
> Project: Flink
>  Issue Type: Sub-task
>  Components: Streaming Connectors
>Reporter: Stephan Ewen
>Assignee: Kostas Kloudas
>Priority: Major
>  Labels: pull-request-available
> Fix For: 1.6.2, 1.7.0
>
>
> S3 offers persistence only when uploads are complete. That means at the end 
> of simple uploads and uploads of parts of a MultiPartUpload.
> We should implement a RecoverableWriter for S3 that does a MultiPartUpload 
> with a Part per checkpoint.
> Recovering the reader needs the MultiPartUploadID and the list of ETags of 
> previous parts.
> We need additional staging of data in Flink state to work around the fact that
>  - Parts in a MultiPartUpload must be at least 5MB
>  - Part sizes must be known up front. (Note that data can still be streamed 
> in the upload)



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-10728) Unexpected items on ASF mirrors

[Chesnay Schepler (JIRA)]

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

Chesnay Schepler commented on FLINK-10728:
--

I simply forgot to send out the announcement mails. Given that Flink itself is 
the only intended user of flink-shaded would an announcement to 
{{d...@flink.apache.org}} be sufficient (as it effectively informs all users)?

That all releases MUST be published on the project website was new to me, will 
add the links shortly.

> Unexpected items on ASF mirrors
> ---
>
> Key: FLINK-10728
> URL: https://issues.apache.org/jira/browse/FLINK-10728
> Project: Flink
>  Issue Type: Bug
> Environment: https://www.apache.org/dist/flink/flink-shaded-4.0/
> https://www.apache.org/dist/flink/flink-shaded-5.0/
>Reporter: Sebb
>Assignee: Chesnay Schepler
>Priority: Major
>
> The flink-shaded artifacts are not linked from the download page, and don't 
> appear to have been announced on any ASF mailing lists, so it's not clear why 
> they are on the ASF mirrors.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Assigned] (FLINK-10728) Unexpected items on ASF mirrors

[Chesnay Schepler (JIRA)]

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

Chesnay Schepler reassigned FLINK-10728:


Assignee: Chesnay Schepler

> Unexpected items on ASF mirrors
> ---
>
> Key: FLINK-10728
> URL: https://issues.apache.org/jira/browse/FLINK-10728
> Project: Flink
>  Issue Type: Bug
> Environment: https://www.apache.org/dist/flink/flink-shaded-4.0/
> https://www.apache.org/dist/flink/flink-shaded-5.0/
>Reporter: Sebb
>Assignee: Chesnay Schepler
>Priority: Major
>
> The flink-shaded artifacts are not linked from the download page, and don't 
> appear to have been announced on any ASF mailing lists, so it's not clear why 
> they are on the ASF mirrors.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-2028) Download pages must not link to snapshots

[ASF GitHub Bot (JIRA)]

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

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

zentol opened a new pull request #131: [FLINK-2028] Remove snapshot link
URL: https://github.com/apache/flink-web/pull/131
 
 
   This PR removes the snapshot build from the download page according to [ASF 
policy](http://www.apache.org/legal/release-policy.html#what).
   
   > Do not include any links on the project website that might encourage 
non-developers to download and use nightly builds, snapshots, release 
candidates, or any other similar package.


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


> Download pages must not link to snapshots
> -
>
> Key: FLINK-2028
> URL: https://issues.apache.org/jira/browse/FLINK-2028
> Project: Flink
>  Issue Type: Bug
>Reporter: Sebb
>Assignee: Ufuk Celebi
>Priority: Major
>  Labels: pull-request-available
>
> Snapshot builds that have not passed a PMC vote must not be advertised to the 
> general public, so must not be included on the downloads page.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Updated] (FLINK-2028) Download pages must not link to snapshots

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot updated FLINK-2028:
--
Labels: pull-request-available  (was: )

> Download pages must not link to snapshots
> -
>
> Key: FLINK-2028
> URL: https://issues.apache.org/jira/browse/FLINK-2028
> Project: Flink
>  Issue Type: Bug
>Reporter: Sebb
>Assignee: Ufuk Celebi
>Priority: Major
>  Labels: pull-request-available
>
> Snapshot builds that have not passed a PMC vote must not be advertised to the 
> general public, so must not be included on the downloads page.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-2028) Download pages must not link to snapshots

[Chesnay Schepler (JIRA)]

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

Chesnay Schepler commented on FLINK-2028:
-

[~s...@apache.org] Is it allowed to link to said snapshot builds from the wiki?

> Download pages must not link to snapshots
> -
>
> Key: FLINK-2028
> URL: https://issues.apache.org/jira/browse/FLINK-2028
> Project: Flink
>  Issue Type: Bug
>Reporter: Sebb
>Assignee: Ufuk Celebi
>Priority: Major
>
> Snapshot builds that have not passed a PMC vote must not be advertised to the 
> general public, so must not be included on the downloads page.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (FLINK-10728) Unexpected items on ASF mirrors

[Sebb (JIRA)]

Sebb created FLINK-10728:


 Summary: Unexpected items on ASF mirrors
 Key: FLINK-10728
 URL: https://issues.apache.org/jira/browse/FLINK-10728
 Project: Flink
  Issue Type: Bug
 Environment: https://www.apache.org/dist/flink/flink-shaded-4.0/
https://www.apache.org/dist/flink/flink-shaded-5.0/
Reporter: Sebb


The flink-shaded artifacts are not linked from the download page, and don't 
appear to have been announced on any ASF mailing lists, so it's not clear why 
they are on the ASF mirrors.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-10628) Add end-to-end test for REST communication encryption

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10628:


azagrebin opened a new pull request #6967: [FLINK-10628][E2E][SSL] Enable 
mutual REST SSL auth in e2e tests
URL: https://github.com/apache/flink/pull/6967
 
 
   ## What is the purpose of the change
   
   This PR enables mutual SSL authentication in batch all around and streaming 
file sink end-to-end tests. It also splits deprecated ssl configs into separate 
SSL configuration for internal and external REST communication. The curl 
commands, which query Flink REST API, are fixed with relevant SSL args to check 
and provide certificates where needed.
   
   ## Brief change log
   
 - separate common_ssl.sh from common.sh
 - fix curls with SSL in common.sh which query Flink
 - enable mutual SSL authentication in test_batch_allround.sh and 
test_streaming_file_sink.sh
   
   ## Verifying this change
   
   build Flink and run:
   FLINK_DIR=./build-target ./flink-end-to-end-tests/run-single-test.sh 
./flink-end-to-end-tests/test-scripts/test_batch_allround.sh
   FLINK_DIR=./build-target ./flink-end-to-end-tests/run-single-test.sh 
./flink-end-to-end-tests/test-scripts/test_streaming_file_sink.sh
   
   ## 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


> Add end-to-end test for REST communication encryption
> -
>
> Key: FLINK-10628
> URL: https://issues.apache.org/jira/browse/FLINK-10628
> Project: Flink
>  Issue Type: Sub-task
>  Components: REST
>Affects Versions: 1.7.0
>Reporter: Till Rohrmann
>Assignee: Andrey Zagrebin
>Priority: Major
>  Labels: pull-request-available
> Fix For: 1.7.0
>
>
> With FLINK-10371 we added support for mutual authentication for the REST 
> communication. We should adapt one of the existing end-to-end tests to 
> require this feature for the REST communication.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Updated] (FLINK-10628) Add end-to-end test for REST communication encryption

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot updated FLINK-10628:
---
Labels: pull-request-available  (was: )

> Add end-to-end test for REST communication encryption
> -
>
> Key: FLINK-10628
> URL: https://issues.apache.org/jira/browse/FLINK-10628
> Project: Flink
>  Issue Type: Sub-task
>  Components: REST
>Affects Versions: 1.7.0
>Reporter: Till Rohrmann
>Assignee: Andrey Zagrebin
>Priority: Major
>  Labels: pull-request-available
> Fix For: 1.7.0
>
>
> With FLINK-10371 we added support for mutual authentication for the REST 
> communication. We should adapt one of the existing end-to-end tests to 
> require this feature for the REST communication.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[GitHub] azagrebin opened a new pull request #6967: [FLINK-10628][E2E][SSL] Enable mutual REST SSL auth in e2e tests

[GitBox]

azagrebin opened a new pull request #6967: [FLINK-10628][E2E][SSL] Enable 
mutual REST SSL auth in e2e tests
URL: https://github.com/apache/flink/pull/6967
 
 
   ## What is the purpose of the change
   
   This PR enables mutual SSL authentication in batch all around and streaming 
file sink end-to-end tests. It also splits deprecated ssl configs into separate 
SSL configuration for internal and external REST communication. The curl 
commands, which query Flink REST API, are fixed with relevant SSL args to check 
and provide certificates where needed.
   
   ## Brief change log
   
 - separate common_ssl.sh from common.sh
 - fix curls with SSL in common.sh which query Flink
 - enable mutual SSL authentication in test_batch_allround.sh and 
test_streaming_file_sink.sh
   
   ## Verifying this change
   
   build Flink and run:
   FLINK_DIR=./build-target ./flink-end-to-end-tests/run-single-test.sh 
./flink-end-to-end-tests/test-scripts/test_batch_allround.sh
   FLINK_DIR=./build-target ./flink-end-to-end-tests/run-single-test.sh 
./flink-end-to-end-tests/test-scripts/test_streaming_file_sink.sh
   
   ## Does this pull request potentially affect one of the following parts:
   
 - Dependencies (does it add or upgrade a dependency): (no)
 - The public API, i.e., is any changed class annotated with 
`@Public(Evolving)`: (no)
 - The serializers: (no)
 - The runtime per-record code paths (performance sensitive): (no)
 - Anything that affects deployment or recovery: JobManager (and its 
components), Checkpointing, Yarn/Mesos, ZooKeeper: (no)
 - The S3 file system connector: (no)
   
   ## Documentation
   
 - Does this pull request introduce a new feature? (no)
 - If yes, how is the feature documented? (not applicable)
   


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[jira] [Commented] (FLINK-10556) Integration with Apache Hive

[Xuefu Zhang (JIRA)]

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

Xuefu Zhang commented on FLINK-10556:
-

I have attached the design doc on Hive metastore integration. It covers using 
Hive as a schema registry as well as a metadata connector for Hive. Please 
review and share your input. Thanks.

> Integration with Apache Hive
> 
>
> Key: FLINK-10556
> URL: https://issues.apache.org/jira/browse/FLINK-10556
> Project: Flink
>  Issue Type: New Feature
>  Components: Batch Connectors and Input/Output Formats, SQL Client, 
> Table API  SQL
>Affects Versions: 1.6.0
>Reporter: Xuefu Zhang
>Assignee: Xuefu Zhang
>Priority: Major
> Attachments: Flink-Hive Metastore Connectivity Design.pdf, Proposal_ 
> Integrate Flink with Hive Ecosystem.pdf
>
>
> This is an umbrella JIRA tracking all enhancement and issues related to 
> integrating Flink with Hive ecosystem. This is an outcome of a discussion in 
> the community, and thanks go to everyone that provided feedback and interest.
> Specifically, we'd like to see the following features and capabilities 
> immediately in Flink:
> # Metadata interoperability
> # Data interoperability
> # Data type compatibility
> # Hive UDF support
> # DDL/DML/Query language compatibility
> For a longer term, we'd also like to add or improve:
> # Compatible SQL service, client tools, JDBC/ODBC drivers
> # Better task failure tolerance and task scheduling
> # Support other user customizations in Hive (storage handlers, serdes, etc).
> I will provide more details regarding the proposal in a doc shortly. Design 
> doc, if deemed necessary, will be provided in each related sub tasks under 
> this JIRA.
> Feedback and contributions are greatly welcome!



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Updated] (FLINK-10556) Integration with Apache Hive

[Xuefu Zhang (JIRA)]

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

Xuefu Zhang updated FLINK-10556:

Attachment: Flink-Hive Metastore Connectivity Design.pdf

> Integration with Apache Hive
> 
>
> Key: FLINK-10556
> URL: https://issues.apache.org/jira/browse/FLINK-10556
> Project: Flink
>  Issue Type: New Feature
>  Components: Batch Connectors and Input/Output Formats, SQL Client, 
> Table API  SQL
>Affects Versions: 1.6.0
>Reporter: Xuefu Zhang
>Assignee: Xuefu Zhang
>Priority: Major
> Attachments: Flink-Hive Metastore Connectivity Design.pdf, Proposal_ 
> Integrate Flink with Hive Ecosystem.pdf
>
>
> This is an umbrella JIRA tracking all enhancement and issues related to 
> integrating Flink with Hive ecosystem. This is an outcome of a discussion in 
> the community, and thanks go to everyone that provided feedback and interest.
> Specifically, we'd like to see the following features and capabilities 
> immediately in Flink:
> # Metadata interoperability
> # Data interoperability
> # Data type compatibility
> # Hive UDF support
> # DDL/DML/Query language compatibility
> For a longer term, we'd also like to add or improve:
> # Compatible SQL service, client tools, JDBC/ODBC drivers
> # Better task failure tolerance and task scheduling
> # Support other user customizations in Hive (storage handlers, serdes, etc).
> I will provide more details regarding the proposal in a doc shortly. Design 
> doc, if deemed necessary, will be provided in each related sub tasks under 
> this JIRA.
> Feedback and contributions are greatly welcome!



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Reopened] (FLINK-2028) Download pages must not link to snapshots

[Sebb (JIRA)]

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

Sebb reopened FLINK-2028:
-

The download page contains a link to a SNAPSHOT.
This is not allowed on a public download page.
Please remove the link ASAP

> Download pages must not link to snapshots
> -
>
> Key: FLINK-2028
> URL: https://issues.apache.org/jira/browse/FLINK-2028
> Project: Flink
>  Issue Type: Bug
>Reporter: Sebb
>Assignee: Ufuk Celebi
>Priority: Major
>
> Snapshot builds that have not passed a PMC vote must not be advertised to the 
> general public, so must not be included on the downloads page.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[GitHub] kl0u commented on issue #6608: [FLINK-10203]Support truncate method for old Hadoop versions in HadoopRecoverableFsDataOutputStream

[GitBox]

kl0u commented on issue #6608: [FLINK-10203]Support truncate method for old 
Hadoop versions in HadoopRecoverableFsDataOutputStream
URL: https://github.com/apache/flink/pull/6608#issuecomment-434409845
 
 
   Hi Artsem, you are correct that it is not used but I already have a branch
   for it and there is an open Jira for that that I have assigned to myself.
   
   On Tue, Oct 30, 2018, 17:01 Artsem Semianenka 
   wrote:
   
   > @StephanEwen  I really like your idea
   > regarding recoverable writer with "Recover for resume" property. I found
   > the method which you are talking about:
   > boolean supportsResume()
   > in RecoverableWriterm
   > 

   > interface, but as far as I can see this method is not using in the Flink
   > project.
   > Using search by the whole project I found only implementation of this
   > method, but no one invocation of the method
   > 
https://github.com/apache/flink/search?q=supportsResume_q=supportsResume
   >
   > —
   > You are receiving this because you are subscribed to this thread.
   > Reply to this email directly, view it on GitHub
   > , or mute
   > the thread
   > 

   > .
   >
   


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-10203) Support truncate method for old Hadoop versions in HadoopRecoverableFsDataOutputStream

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10203:


kl0u commented on issue #6608: [FLINK-10203]Support truncate method for old 
Hadoop versions in HadoopRecoverableFsDataOutputStream
URL: https://github.com/apache/flink/pull/6608#issuecomment-434409845
 
 
   Hi Artsem, you are correct that it is not used but I already have a branch
   for it and there is an open Jira for that that I have assigned to myself.
   
   On Tue, Oct 30, 2018, 17:01 Artsem Semianenka 
   wrote:
   
   > @StephanEwen  I really like your idea
   > regarding recoverable writer with "Recover for resume" property. I found
   > the method which you are talking about:
   > boolean supportsResume()
   > in RecoverableWriterm
   > 

   > interface, but as far as I can see this method is not using in the Flink
   > project.
   > Using search by the whole project I found only implementation of this
   > method, but no one invocation of the method
   > 
https://github.com/apache/flink/search?q=supportsResume_q=supportsResume
   >
   > —
   > You are receiving this because you are subscribed to this thread.
   > Reply to this email directly, view it on GitHub
   > , or mute
   > the thread
   > 

   > .
   >
   


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 truncate method for old Hadoop versions in 
> HadoopRecoverableFsDataOutputStream
> --
>
> Key: FLINK-10203
> URL: https://issues.apache.org/jira/browse/FLINK-10203
> Project: Flink
>  Issue Type: Bug
>  Components: DataStream API, filesystem-connector
>Affects Versions: 1.6.0, 1.6.1, 1.7.0
>Reporter: Artsem Semianenka
>Assignee: Artsem Semianenka
>Priority: Major
>  Labels: pull-request-available
> Attachments: legacy truncate logic.pdf
>
>
> New StreamingFileSink ( introduced in 1.6 Flink version ) use 
> HadoopRecoverableFsDataOutputStream wrapper to write data in HDFS.
> HadoopRecoverableFsDataOutputStream is a wrapper for FSDataOutputStream to 
> have an ability to restore from certain point of file after failure and 
> continue write data. To achieve this recover functionality the 
> HadoopRecoverableFsDataOutputStream use "truncate" method which was 
> introduced only in Hadoop 2.7 .
> Unfortunately there are a few official Hadoop distributive which latest 
> version still use Hadoop 2.6 (This distributives: Cloudera, Pivotal HD ). As 
> the result Flinks Hadoop connector can't work with this distributives.
> Flink declares that supported Hadoop from version 2.4.0 upwards 
> ([https://ci.apache.org/projects/flink/flink-docs-release-1.6/start/building.html#hadoop-versions])
> I guess we should emulate the functionality of "truncate" method for older 
> Hadoop versions.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Closed] (FLINK-9635) Local recovery scheduling can cause spread out of tasks

[Stefan Richter (JIRA)]

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

Stefan Richter closed FLINK-9635.
-
Resolution: Fixed

Merged in:
master: c8a6471a58

> Local recovery scheduling can cause spread out of tasks
> ---
>
> Key: FLINK-9635
> URL: https://issues.apache.org/jira/browse/FLINK-9635
> Project: Flink
>  Issue Type: Bug
>  Components: Distributed Coordination
>Affects Versions: 1.5.0, 1.6.2
>Reporter: Till Rohrmann
>Assignee: Stefan Richter
>Priority: Critical
>  Labels: pull-request-available
> Fix For: 1.7.0
>
>
> In order to make local recovery work, Flink's scheduling was changed such 
> that it tries to be rescheduled to its previous location. In order to not 
> occupy slots which have state of other tasks cached, the strategy will 
> request a new slot if the old slot identified by the previous allocation id 
> is no longer present. This also applies to newly allocated slots because 
> there is no distinction between new or already used. This behaviour can cause 
> that every tasks gets deployed to its own slot if the {{SlotPool}} has 
> released all slots in the meantime, for example. The consequence could be 
> that a job can no longer be executed after a failure because it needs more 
> slots than before.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-9635) Local recovery scheduling can cause spread out of tasks

[ASF GitHub Bot (JIRA)]

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

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

asfgit closed pull request #6961: [FLINK-9635] Fix scheduling for local recovery
URL: https://github.com/apache/flink/pull/6961
 
 
   

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/clusterframework/types/SlotProfile.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/SlotProfile.java
index 7cb364d687f..ab682a02dd1 100644
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/SlotProfile.java
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/SlotProfile.java
@@ -25,6 +25,7 @@
 
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Set;
 
 /**
  * A slot profile describes the profile of a slot into which a task wants to 
be scheduled. The profile contains
@@ -47,16 +48,30 @@
 
/** This contains desired allocation ids of the slot. */
@Nonnull
-   private final Collection priorAllocations;
+   private final Collection preferredAllocations;
+
+   /** This contains all prior allocation ids from the whole execution 
graph. */
+   @Nonnull
+   private final Set previousExecutionGraphAllocations;
+
+   public SlotProfile(
+   @Nonnull ResourceProfile resourceProfile,
+   @Nonnull Collection preferredLocations,
+   @Nonnull Collection preferredAllocations) {
+
+   this(resourceProfile, preferredLocations, preferredAllocations, 
Collections.emptySet());
+   }
 
public SlotProfile(
@Nonnull ResourceProfile resourceProfile,
@Nonnull Collection preferredLocations,
-   @Nonnull Collection priorAllocations) {
+   @Nonnull Collection preferredAllocations,
+   @Nonnull Set previousExecutionGraphAllocations) {
 
this.resourceProfile = resourceProfile;
this.preferredLocations = preferredLocations;
-   this.priorAllocations = priorAllocations;
+   this.preferredAllocations = preferredAllocations;
+   this.previousExecutionGraphAllocations = 
previousExecutionGraphAllocations;
}
 
/**
@@ -79,8 +94,18 @@ public ResourceProfile getResourceProfile() {
 * Returns the desired allocation ids for the slot.
 */
@Nonnull
-   public Collection getPriorAllocations() {
-   return priorAllocations;
+   public Collection getPreferredAllocations() {
+   return preferredAllocations;
+   }
+
+   /**
+* Returns a set of all previous allocation ids from the execution 
graph.
+*
+* This is optional and can be empty if unused.
+*/
+   @Nonnull
+   public Set getPreviousExecutionGraphAllocations() {
+   return previousExecutionGraphAllocations;
}
 
/**
diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
index 801f35a41dc..cbf51037b8b 100644
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
@@ -58,6 +58,7 @@
 
 import org.slf4j.Logger;
 
+import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
 import java.util.ArrayList;
@@ -65,6 +66,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ConcurrentLinkedQueue;
@@ -385,7 +387,8 @@ public void setInitialState(@Nullable JobManagerTaskRestore 
taskRestore) {
return scheduleForExecution(
resourceProvider,
allowQueued,
-   LocationPreferenceConstraint.ANY);
+   LocationPreferenceConstraint.ANY,
+   Collections.emptySet());
}
 
/**
@@ -397,18 +400,22 @@ public void setInitialState(@Nullable 
JobManagerTaskRestore taskRestore) {
 * @param queued Flag to indicate whether the scheduler may queue this 
task if it cannot
 *   immediately deploy it.
 * @param locationPreferenceConstraint constraint for the location 
preferences
+* @param 

[GitHub] asfgit closed pull request #6961: [FLINK-9635] Fix scheduling for local recovery

[GitBox]

asfgit closed pull request #6961: [FLINK-9635] Fix scheduling for local recovery
URL: https://github.com/apache/flink/pull/6961
 
 
   

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/clusterframework/types/SlotProfile.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/SlotProfile.java
index 7cb364d687f..ab682a02dd1 100644
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/SlotProfile.java
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/SlotProfile.java
@@ -25,6 +25,7 @@
 
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Set;
 
 /**
  * A slot profile describes the profile of a slot into which a task wants to 
be scheduled. The profile contains
@@ -47,16 +48,30 @@
 
/** This contains desired allocation ids of the slot. */
@Nonnull
-   private final Collection priorAllocations;
+   private final Collection preferredAllocations;
+
+   /** This contains all prior allocation ids from the whole execution 
graph. */
+   @Nonnull
+   private final Set previousExecutionGraphAllocations;
+
+   public SlotProfile(
+   @Nonnull ResourceProfile resourceProfile,
+   @Nonnull Collection preferredLocations,
+   @Nonnull Collection preferredAllocations) {
+
+   this(resourceProfile, preferredLocations, preferredAllocations, 
Collections.emptySet());
+   }
 
public SlotProfile(
@Nonnull ResourceProfile resourceProfile,
@Nonnull Collection preferredLocations,
-   @Nonnull Collection priorAllocations) {
+   @Nonnull Collection preferredAllocations,
+   @Nonnull Set previousExecutionGraphAllocations) {
 
this.resourceProfile = resourceProfile;
this.preferredLocations = preferredLocations;
-   this.priorAllocations = priorAllocations;
+   this.preferredAllocations = preferredAllocations;
+   this.previousExecutionGraphAllocations = 
previousExecutionGraphAllocations;
}
 
/**
@@ -79,8 +94,18 @@ public ResourceProfile getResourceProfile() {
 * Returns the desired allocation ids for the slot.
 */
@Nonnull
-   public Collection getPriorAllocations() {
-   return priorAllocations;
+   public Collection getPreferredAllocations() {
+   return preferredAllocations;
+   }
+
+   /**
+* Returns a set of all previous allocation ids from the execution 
graph.
+*
+* This is optional and can be empty if unused.
+*/
+   @Nonnull
+   public Set getPreviousExecutionGraphAllocations() {
+   return previousExecutionGraphAllocations;
}
 
/**
diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
index 801f35a41dc..cbf51037b8b 100644
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
@@ -58,6 +58,7 @@
 
 import org.slf4j.Logger;
 
+import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
 import java.util.ArrayList;
@@ -65,6 +66,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ConcurrentLinkedQueue;
@@ -385,7 +387,8 @@ public void setInitialState(@Nullable JobManagerTaskRestore 
taskRestore) {
return scheduleForExecution(
resourceProvider,
allowQueued,
-   LocationPreferenceConstraint.ANY);
+   LocationPreferenceConstraint.ANY,
+   Collections.emptySet());
}
 
/**
@@ -397,18 +400,22 @@ public void setInitialState(@Nullable 
JobManagerTaskRestore taskRestore) {
 * @param queued Flag to indicate whether the scheduler may queue this 
task if it cannot
 *   immediately deploy it.
 * @param locationPreferenceConstraint constraint for the location 
preferences
+* @param allPreviousExecutionGraphAllocationIds set with all previous 
allocation ids in the job graph.
+* Can be empty if the 
allocation ids are not required for scheduling.
 * @return Future which is completed once 

[jira] [Commented] (FLINK-9635) Local recovery scheduling can cause spread out of tasks

[ASF GitHub Bot (JIRA)]

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

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

StefanRRichter commented on issue #6961: [FLINK-9635] Fix scheduling for local 
recovery
URL: https://github.com/apache/flink/pull/6961#issuecomment-434388623
 
 
   Thanks for the review @tillrohrmann! Travis looks good, proceeding with the 
merge.


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


> Local recovery scheduling can cause spread out of tasks
> ---
>
> Key: FLINK-9635
> URL: https://issues.apache.org/jira/browse/FLINK-9635
> Project: Flink
>  Issue Type: Bug
>  Components: Distributed Coordination
>Affects Versions: 1.5.0, 1.6.2
>Reporter: Till Rohrmann
>Assignee: Stefan Richter
>Priority: Critical
>  Labels: pull-request-available
> Fix For: 1.7.0
>
>
> In order to make local recovery work, Flink's scheduling was changed such 
> that it tries to be rescheduled to its previous location. In order to not 
> occupy slots which have state of other tasks cached, the strategy will 
> request a new slot if the old slot identified by the previous allocation id 
> is no longer present. This also applies to newly allocated slots because 
> there is no distinction between new or already used. This behaviour can cause 
> that every tasks gets deployed to its own slot if the {{SlotPool}} has 
> released all slots in the meantime, for example. The consequence could be 
> that a job can no longer be executed after a failure because it needs more 
> slots than before.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[GitHub] StefanRRichter commented on issue #6961: [FLINK-9635] Fix scheduling for local recovery

[GitBox]

StefanRRichter commented on issue #6961: [FLINK-9635] Fix scheduling for local 
recovery
URL: https://github.com/apache/flink/pull/6961#issuecomment-434388623
 
 
   Thanks for the review @tillrohrmann! Travis looks good, proceeding with the 
merge.


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-10166) Dependency problems when executing SQL query in sql-client

[Timo Walther (JIRA)]

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

Timo Walther reassigned FLINK-10166:


Assignee: Timo Walther

> Dependency problems when executing SQL query in sql-client
> --
>
> Key: FLINK-10166
> URL: https://issues.apache.org/jira/browse/FLINK-10166
> Project: Flink
>  Issue Type: Bug
>  Components: Table API  SQL
>Affects Versions: 1.6.0
>Reporter: Dawid Wysakowicz
>Assignee: Timo Walther
>Priority: Blocker
>  Labels: pull-request-available
> Fix For: 1.7.0
>
>
> When tried to run query:
> {code}
> select count(distinct name) from (Values ('a'), ('b')) AS NameTable(name)
> {code}
> in {{sql-client.sh}} I got:
> {code}
> [ERROR] Could not execute SQL statement. Reason:
> org.codehaus.commons.compiler.CompileException: Line 43, Column 10: Unknown 
> variable or type "org.apache.commons.codec.binary.Base64"
> {code}



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-10166) Dependency problems when executing SQL query in sql-client

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10166:


twalthr opened a new pull request #6966: [FLINK-10166] [table] Reduce 
dependencies by removing org.apache.commons
URL: https://github.com/apache/flink/pull/6966
 
 
   ## What is the purpose of the change
   
   This PR removes all dependencies to `org.apache.commons` libraries. In the 
past we only used a couple of methods that were partially pulled in from Hadoop 
causing the issues mentioned in the JIRA ticket.
   
   ## Brief change log
   
   - Add more utility method to the Flink code base in order to reduce external 
dependencies
   
   ## Verifying this change
   
   This change is already covered by existing tests.
   I added additional test such as `EncodingUtilsTest` and `TypeCheckUtilsTest`.
   
   ## Does this pull request potentially affect one of the following parts:
   
 - Dependencies (does it add or upgrade a dependency): yes
 - 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


> Dependency problems when executing SQL query in sql-client
> --
>
> Key: FLINK-10166
> URL: https://issues.apache.org/jira/browse/FLINK-10166
> Project: Flink
>  Issue Type: Bug
>  Components: Table API  SQL
>Affects Versions: 1.6.0
>Reporter: Dawid Wysakowicz
>Priority: Blocker
>  Labels: pull-request-available
> Fix For: 1.7.0
>
>
> When tried to run query:
> {code}
> select count(distinct name) from (Values ('a'), ('b')) AS NameTable(name)
> {code}
> in {{sql-client.sh}} I got:
> {code}
> [ERROR] Could not execute SQL statement. Reason:
> org.codehaus.commons.compiler.CompileException: Line 43, Column 10: Unknown 
> variable or type "org.apache.commons.codec.binary.Base64"
> {code}



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[GitHub] twalthr opened a new pull request #6966: [FLINK-10166] [table] Reduce dependencies by removing org.apache.commons

[GitBox]

twalthr opened a new pull request #6966: [FLINK-10166] [table] Reduce 
dependencies by removing org.apache.commons
URL: https://github.com/apache/flink/pull/6966
 
 
   ## What is the purpose of the change
   
   This PR removes all dependencies to `org.apache.commons` libraries. In the 
past we only used a couple of methods that were partially pulled in from Hadoop 
causing the issues mentioned in the JIRA ticket.
   
   ## Brief change log
   
   - Add more utility method to the Flink code base in order to reduce external 
dependencies
   
   ## Verifying this change
   
   This change is already covered by existing tests.
   I added additional test such as `EncodingUtilsTest` and `TypeCheckUtilsTest`.
   
   ## Does this pull request potentially affect one of the following parts:
   
 - Dependencies (does it add or upgrade a dependency): yes
 - The public API, i.e., is any changed class annotated with 
`@Public(Evolving)`: no
 - The serializers: no
 - The runtime per-record code paths (performance sensitive): no
 - Anything that affects deployment or recovery: JobManager (and its 
components), Checkpointing, Yarn/Mesos, ZooKeeper: no
 - The S3 file system connector: no
   
   ## Documentation
   
 - Does this pull request introduce a new feature? no
 - If yes, how is the feature documented? not applicable
   


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[jira] [Commented] (FLINK-10726) Streaming SQL end-to-end test fails due to recent flink-table-common changes

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10726:


asfgit closed pull request #6964: [FLINK-10726] [table] Include 
flink-table-common in flink-table jar
URL: https://github.com/apache/flink/pull/6964
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/flink-libraries/flink-table/pom.xml 
b/flink-libraries/flink-table/pom.xml
index 536f9c9b4d4..14653710b4e 100644
--- a/flink-libraries/flink-table/pom.xml
+++ b/flink-libraries/flink-table/pom.xml
@@ -282,6 +282,7 @@ under the License.

commons-lang:*
 

+   
org.apache.flink:flink-table-common

commons-codec:*

org.apache.commons:commons-lang3

org.codehaus.janino:*


 


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


> Streaming SQL end-to-end test fails due to recent flink-table-common changes
> 
>
> Key: FLINK-10726
> URL: https://issues.apache.org/jira/browse/FLINK-10726
> Project: Flink
>  Issue Type: Bug
>  Components: E2E Tests, Table API  SQL
>Affects Versions: 1.7.0
>Reporter: Timo Walther
>Assignee: Timo Walther
>Priority: Major
>  Labels: pull-request-available
> Fix For: 1.7.0
>
>
> The streaming SQL end-to-end test fails with:
> {code}
> java.lang.NoClassDefFoundError: org/apache/flink/table/api/TableException
>   at 
> org.apache.flink.sql.tests.StreamSQLTestProgram.main(StreamSQLTestProgram.java:85)
>   at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>   at 
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
>   at 
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
>   at java.lang.reflect.Method.invoke(Method.java:498)
>   at 
> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:529)
>   at 
> org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:421)
>   at 
> org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:427)
>   at 
> org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:813)
>   at 
> org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:287)
>   at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:213)
>   at 
> org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1050)
>   at 
> org.apache.flink.client.cli.CliFrontend.lambda$main$11(CliFrontend.java:1126)
>   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:1556)
>   at 
> org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41)
>   at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1126)
> Caused by: java.lang.ClassNotFoundException: 
> org.apache.flink.table.api.TableException
>   at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
>   at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
>   at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
>   at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
>   ... 18 more
> {code}
> This has to do with the recent dependency changes for introducing a 
> {{flink-table-common}} module.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[GitHub] asfgit closed pull request #6964: [FLINK-10726] [table] Include flink-table-common in flink-table jar

[GitBox]

asfgit closed pull request #6964: [FLINK-10726] [table] Include 
flink-table-common in flink-table jar
URL: https://github.com/apache/flink/pull/6964
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/flink-libraries/flink-table/pom.xml 
b/flink-libraries/flink-table/pom.xml
index 536f9c9b4d4..14653710b4e 100644
--- a/flink-libraries/flink-table/pom.xml
+++ b/flink-libraries/flink-table/pom.xml
@@ -282,6 +282,7 @@ under the License.

commons-lang:*
 

+   
org.apache.flink:flink-table-common

commons-codec:*

org.apache.commons:commons-lang3

org.codehaus.janino:*


 


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 commented on issue #6924: [FLINK-10600] Provide End-to-end test cases for modern Kafka connectors

[GitBox]

yanghua commented on issue #6924: [FLINK-10600] Provide End-to-end test cases 
for modern Kafka connectors
URL: https://github.com/apache/flink/pull/6924#issuecomment-434372198
 
 
   @pnowojski I just tested again, added `set -x` on the top of 
`kafka-common.sh`, `test_streaming_kafka_common.sh` and 
`test_streaming_kafka.sh`. Everything is OK and it shown `[PASS]`


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-10600) Provide End-to-end test cases for modern Kafka connectors

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10600:


yanghua commented on issue #6924: [FLINK-10600] Provide End-to-end test cases 
for modern Kafka connectors
URL: https://github.com/apache/flink/pull/6924#issuecomment-434372198
 
 
   @pnowojski I just tested again, added `set -x` on the top of 
`kafka-common.sh`, `test_streaming_kafka_common.sh` and 
`test_streaming_kafka.sh`. Everything is OK and it shown `[PASS]`


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


> Provide End-to-end test cases for modern Kafka connectors
> -
>
> Key: FLINK-10600
> URL: https://issues.apache.org/jira/browse/FLINK-10600
> Project: Flink
>  Issue Type: Sub-task
>  Components: Kafka Connector
>Reporter: vinoyang
>Assignee: vinoyang
>Priority: Blocker
>  Labels: pull-request-available
> Fix For: 1.7.0
>
>




--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Resolved] (FLINK-10726) Streaming SQL end-to-end test fails due to recent flink-table-common changes

[Timo Walther (JIRA)]

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

Timo Walther resolved FLINK-10726.
--
   Resolution: Fixed
Fix Version/s: 1.7.0

Fixed in 1.7.0: 3770ac73dcc0b348f17fed98db921b52d87a067c

> Streaming SQL end-to-end test fails due to recent flink-table-common changes
> 
>
> Key: FLINK-10726
> URL: https://issues.apache.org/jira/browse/FLINK-10726
> Project: Flink
>  Issue Type: Bug
>  Components: E2E Tests, Table API  SQL
>Affects Versions: 1.7.0
>Reporter: Timo Walther
>Assignee: Timo Walther
>Priority: Major
>  Labels: pull-request-available
> Fix For: 1.7.0
>
>
> The streaming SQL end-to-end test fails with:
> {code}
> java.lang.NoClassDefFoundError: org/apache/flink/table/api/TableException
>   at 
> org.apache.flink.sql.tests.StreamSQLTestProgram.main(StreamSQLTestProgram.java:85)
>   at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>   at 
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
>   at 
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
>   at java.lang.reflect.Method.invoke(Method.java:498)
>   at 
> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:529)
>   at 
> org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:421)
>   at 
> org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:427)
>   at 
> org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:813)
>   at 
> org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:287)
>   at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:213)
>   at 
> org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1050)
>   at 
> org.apache.flink.client.cli.CliFrontend.lambda$main$11(CliFrontend.java:1126)
>   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:1556)
>   at 
> org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41)
>   at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1126)
> Caused by: java.lang.ClassNotFoundException: 
> org.apache.flink.table.api.TableException
>   at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
>   at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
>   at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
>   at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
>   ... 18 more
> {code}
> This has to do with the recent dependency changes for introducing a 
> {{flink-table-common}} module.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-10632) Run general purpose test job with failures in per-job mode

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10632:


tillrohrmann commented on a change in pull request #6943: [FLINK-10632][e2e] 
Running general purpose testing job with failure in per-job mode
URL: https://github.com/apache/flink/pull/6943#discussion_r229352991
 
 

 ##
 File path: 
flink-end-to-end-tests/test-scripts/test_ha_per_job_cluster_datastream.sh
 ##
 @@ -0,0 +1,160 @@
+#!/usr/bin/env bash
+
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+source "$(dirname "$0")"/common.sh
+source "$(dirname "$0")"/common_ha.sh
+
+TEST_PROGRAM_JAR_NAME=DataStreamAllroundTestProgram.jar
+TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-datastream-allround-test/target/${TEST_PROGRAM_JAR_NAME}
+FLINK_LIB_DIR=${FLINK_DIR}/lib
+JOB_ID=""
+
+function ha_cleanup() {
+  # don't call ourselves again for another signal interruption
+  trap "exit -1" INT
+  # don't call ourselves again for normal exit
+  trap "" EXIT
+
+  stop_watchdogs
+  kill_all 'StandaloneJobClusterEntryPoint'
+  shutdown_all
+  rm ${FLINK_LIB_DIR}/${TEST_PROGRAM_JAR_NAME}
+}
+
+trap ha_cleanup INT
+trap ha_cleanup EXIT
+
+function run_job() {
+local PARALLELISM=$1
+local BACKEND=$2
+local ASYNC=$3
+local INCREM=$4
+
+local CHECKPOINT_DIR="${TEST_DATA_DIR}/checkpoints/"
+
+${FLINK_DIR}/bin/standalone-job.sh start \
+--job-classname 
org.apache.flink.streaming.tests.DataStreamAllroundTestProgram \
+-p ${PARALLELISM} \
 
 Review comment:
   Is this really needed? I think that setting `environment.parallelism` should 
already be enough.


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


> Run general purpose test job with failures in per-job mode
> --
>
> Key: FLINK-10632
> URL: https://issues.apache.org/jira/browse/FLINK-10632
> Project: Flink
>  Issue Type: Sub-task
>  Components: E2E Tests
>Affects Versions: 1.7.0
>Reporter: Till Rohrmann
>Assignee: Dawid Wysakowicz
>Priority: Major
>  Labels: pull-request-available
> Fix For: 1.7.0
>
>
> Similar to FLINK-8973, we should add an end-to-end which runs the general 
> datastream job with failures on a per-job cluster with HA enabled (either 
> directly the {{StandaloneJobClusterEntrypoint}} or a docker image based on 
> this entrypoint).
> We should kill the TMs as well as the cluster entrypoint and verify that the 
> job recovers.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-10632) Run general purpose test job with failures in per-job mode

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10632:


tillrohrmann commented on a change in pull request #6943: [FLINK-10632][e2e] 
Running general purpose testing job with failure in per-job mode
URL: https://github.com/apache/flink/pull/6943#discussion_r229343984
 
 

 ##
 File path: flink-end-to-end-tests/test-scripts/test_ha_datastream.sh
 ##
 @@ -22,6 +22,20 @@ source "$(dirname "$0")"/common_ha.sh
 
 
TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-datastream-allround-test/target/DataStreamAllroundTestProgram.jar
 
+function ha_cleanup() {
+  # don't call ourselves again for another signal interruption
+  trap "exit -1" INT
+  # don't call ourselves again for normal exit
+  trap "" EXIT
+
+  # kill the cluster and zookeeper
+  stop_watchdogs
+  shutdown_all
 
 Review comment:
   Same here.


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


> Run general purpose test job with failures in per-job mode
> --
>
> Key: FLINK-10632
> URL: https://issues.apache.org/jira/browse/FLINK-10632
> Project: Flink
>  Issue Type: Sub-task
>  Components: E2E Tests
>Affects Versions: 1.7.0
>Reporter: Till Rohrmann
>Assignee: Dawid Wysakowicz
>Priority: Major
>  Labels: pull-request-available
> Fix For: 1.7.0
>
>
> Similar to FLINK-8973, we should add an end-to-end which runs the general 
> datastream job with failures on a per-job cluster with HA enabled (either 
> directly the {{StandaloneJobClusterEntrypoint}} or a docker image based on 
> this entrypoint).
> We should kill the TMs as well as the cluster entrypoint and verify that the 
> job recovers.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Comment Edited] (FLINK-10712) RestartPipelinedRegionStrategy does not restore state

[Yun Tang (JIRA)]

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

Yun Tang edited comment on FLINK-10712 at 10/30/18 4:18 PM:


We have already refactored _FailoverRegion.java_ to support fail-over with 
state when using region-failover strategy in our production environment. I'll 
organize related code to crate a new PR in these days.


was (Author: yunta):
We have refactored _FailoverRegion.java_ to support fail-over with state when 
using region-failover strategy. I'll organize related code to crate a new PR in 
these days.

> RestartPipelinedRegionStrategy does not restore state
> -
>
> Key: FLINK-10712
> URL: https://issues.apache.org/jira/browse/FLINK-10712
> Project: Flink
>  Issue Type: Bug
>  Components: Distributed Coordination
>Affects Versions: 1.3.3, 1.4.2, 1.5.5, 1.6.2, 1.7.0
>Reporter: Stefan Richter
>Priority: Critical
> Fix For: 1.7.0
>
>
> RestartPipelinedRegionStrategy does not perform any state restore. This is 
> big problem because all restored regions will be restarted with empty state. 
> We need to take checkpoints into account when restoring.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-10632) Run general purpose test job with failures in per-job mode

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10632:


tillrohrmann commented on a change in pull request #6943: [FLINK-10632][e2e] 
Running general purpose testing job with failure in per-job mode
URL: https://github.com/apache/flink/pull/6943#discussion_r229353255
 
 

 ##
 File path: 
flink-end-to-end-tests/test-scripts/test_ha_per_job_cluster_datastream.sh
 ##
 @@ -0,0 +1,160 @@
+#!/usr/bin/env bash
+
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+source "$(dirname "$0")"/common.sh
+source "$(dirname "$0")"/common_ha.sh
+
+TEST_PROGRAM_JAR_NAME=DataStreamAllroundTestProgram.jar
+TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-datastream-allround-test/target/${TEST_PROGRAM_JAR_NAME}
+FLINK_LIB_DIR=${FLINK_DIR}/lib
+JOB_ID=""
+
+function ha_cleanup() {
+  # don't call ourselves again for another signal interruption
+  trap "exit -1" INT
+  # don't call ourselves again for normal exit
+  trap "" EXIT
+
+  stop_watchdogs
+  kill_all 'StandaloneJobClusterEntryPoint'
+  shutdown_all
+  rm ${FLINK_LIB_DIR}/${TEST_PROGRAM_JAR_NAME}
+}
+
+trap ha_cleanup INT
+trap ha_cleanup EXIT
+
+function run_job() {
+local PARALLELISM=$1
+local BACKEND=$2
+local ASYNC=$3
+local INCREM=$4
+
+local CHECKPOINT_DIR="${TEST_DATA_DIR}/checkpoints/"
+
+${FLINK_DIR}/bin/standalone-job.sh start \
+--job-classname 
org.apache.flink.streaming.tests.DataStreamAllroundTestProgram \
+-p ${PARALLELISM} \
+--environment.parallelism ${PARALLELISM} \
+--test.semantics exactly-once \
+--test.simulate_failure true \
+--test.simulate_failure.num_records 200 \
+--test.simulate_failure.num_checkpoints 1 \
+--test.simulate_failure.max_failures 20 \
+--state_backend ${BACKEND} \
+--state_backend.checkpoint_directory "file://${CHECKPOINT_DIR}" \
+--state_backend.file.async ${ASYNC} \
+--state_backend.rocks.incremental ${INCREM} \
+--sequence_generator_source.sleep_time 15 \
+--sequence_generator_source.sleep_after_elements 1
+}
+
+function verify_logs_per_job() {
+local OUTPUT=$FLINK_DIR/log/*.out
+local JM_FAILURES=$1
+local EXIT_CODE=0
+local VERIFY_CHECKPOINTS=$2
+
+# verify that we have no alerts
+if ! [ `cat ${OUTPUT} | wc -l` -eq 0 ]; then
+echo "FAILURE: Alerts found at the general purpose job."
+EXIT_CODE=1
+fi
+
+# checks that all apart from the first JM recover the failed jobgraph.
+if ! [ `grep -r --include '*standalonejob*.log' 'Found 0 checkpoints in 
ZooKeeper' "${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq 1 ]; then
+echo "FAILURE: A JM did not take over, but started new job."
+EXIT_CODE=1
+fi
+
+if ! [ `grep -r --include '*standalonejob*.log' -P 'Found \d+ checkpoints 
in ZooKeeper' "${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq 
$((JM_FAILURES + 1)) ]; then
+echo "FAILURE: A JM did not take over."
+EXIT_CODE=1
+fi
+
+if [ "$VERIFY_CHECKPOINTS" = true ]; then
+# search the logs for JMs that log completed checkpoints
+if ! [ `grep -r --include '*standalonejob*.log' 'Completed checkpoint' 
"${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq $((JM_FAILURES + 1)) 
]; then
+echo "FAILURE: A JM did not execute the job."
+EXIT_CODE=1
+fi
+fi
+
+if [[ $EXIT_CODE != 0 ]]; then
+echo "One or more tests FAILED."
+exit $EXIT_CODE
+fi
+}
+
+function run_ha_test() {
+local PARALLELISM=$1
+local BACKEND=$2
+local ASYNC=$3
+local INCREM=$4
+
+local JM_KILLS=3
+
+
+CLEARED=0
+
+# add job jar to cluster classpath
+cp ${TEST_PROGRAM_JAR} ${FLINK_LIB_DIR}
+
+# start the cluster on HA mode
+create_ha_config
+start_local_zk
+
+echo "Running on HA mode: 

[jira] [Commented] (FLINK-10632) Run general purpose test job with failures in per-job mode

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10632:


tillrohrmann commented on a change in pull request #6943: [FLINK-10632][e2e] 
Running general purpose testing job with failure in per-job mode
URL: https://github.com/apache/flink/pull/6943#discussion_r229346067
 
 

 ##
 File path: 
flink-end-to-end-tests/test-scripts/test_ha_per_job_cluster_datastream.sh
 ##
 @@ -0,0 +1,160 @@
+#!/usr/bin/env bash
+
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+source "$(dirname "$0")"/common.sh
+source "$(dirname "$0")"/common_ha.sh
+
+TEST_PROGRAM_JAR_NAME=DataStreamAllroundTestProgram.jar
+TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-datastream-allround-test/target/${TEST_PROGRAM_JAR_NAME}
+FLINK_LIB_DIR=${FLINK_DIR}/lib
+JOB_ID=""
+
+function ha_cleanup() {
+  # don't call ourselves again for another signal interruption
+  trap "exit -1" INT
+  # don't call ourselves again for normal exit
+  trap "" EXIT
+
+  stop_watchdogs
+  kill_all 'StandaloneJobClusterEntryPoint'
+  shutdown_all
+  rm ${FLINK_LIB_DIR}/${TEST_PROGRAM_JAR_NAME}
+}
+
+trap ha_cleanup INT
+trap ha_cleanup EXIT
+
+function run_job() {
+local PARALLELISM=$1
+local BACKEND=$2
+local ASYNC=$3
+local INCREM=$4
+
+local CHECKPOINT_DIR="${TEST_DATA_DIR}/checkpoints/"
+
+${FLINK_DIR}/bin/standalone-job.sh start \
+--job-classname 
org.apache.flink.streaming.tests.DataStreamAllroundTestProgram \
+-p ${PARALLELISM} \
+--environment.parallelism ${PARALLELISM} \
+--test.semantics exactly-once \
+--test.simulate_failure true \
+--test.simulate_failure.num_records 200 \
+--test.simulate_failure.num_checkpoints 1 \
+--test.simulate_failure.max_failures 20 \
+--state_backend ${BACKEND} \
+--state_backend.checkpoint_directory "file://${CHECKPOINT_DIR}" \
+--state_backend.file.async ${ASYNC} \
+--state_backend.rocks.incremental ${INCREM} \
+--sequence_generator_source.sleep_time 15 \
+--sequence_generator_source.sleep_after_elements 1
+}
+
+function verify_logs_per_job() {
+local OUTPUT=$FLINK_DIR/log/*.out
+local JM_FAILURES=$1
+local EXIT_CODE=0
+local VERIFY_CHECKPOINTS=$2
+
+# verify that we have no alerts
+if ! [ `cat ${OUTPUT} | wc -l` -eq 0 ]; then
+echo "FAILURE: Alerts found at the general purpose job."
+EXIT_CODE=1
+fi
+
+# checks that all apart from the first JM recover the failed jobgraph.
+if ! [ `grep -r --include '*standalonejob*.log' 'Found 0 checkpoints in 
ZooKeeper' "${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq 1 ]; then
+echo "FAILURE: A JM did not take over, but started new job."
+EXIT_CODE=1
+fi
+
+if ! [ `grep -r --include '*standalonejob*.log' -P 'Found \d+ checkpoints 
in ZooKeeper' "${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq 
$((JM_FAILURES + 1)) ]; then
+echo "FAILURE: A JM did not take over."
+EXIT_CODE=1
+fi
+
+if [ "$VERIFY_CHECKPOINTS" = true ]; then
+# search the logs for JMs that log completed checkpoints
+if ! [ `grep -r --include '*standalonejob*.log' 'Completed checkpoint' 
"${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq $((JM_FAILURES + 1)) 
]; then
+echo "FAILURE: A JM did not execute the job."
+EXIT_CODE=1
+fi
+fi
+
+if [[ $EXIT_CODE != 0 ]]; then
+echo "One or more tests FAILED."
+exit $EXIT_CODE
+fi
+}
+
+function run_ha_test() {
+local PARALLELISM=$1
+local BACKEND=$2
+local ASYNC=$3
+local INCREM=$4
+
+local JM_KILLS=3
+
+
+CLEARED=0
+
+# add job jar to cluster classpath
+cp ${TEST_PROGRAM_JAR} ${FLINK_LIB_DIR}
+
+# start the cluster on HA mode
+create_ha_config
+start_local_zk
+
+echo "Running on HA mode: 

[jira] [Commented] (FLINK-10632) Run general purpose test job with failures in per-job mode

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10632:


tillrohrmann commented on a change in pull request #6943: [FLINK-10632][e2e] 
Running general purpose testing job with failure in per-job mode
URL: https://github.com/apache/flink/pull/6943#discussion_r229358923
 
 

 ##
 File path: 
flink-end-to-end-tests/test-scripts/test_ha_per_job_cluster_datastream.sh
 ##
 @@ -0,0 +1,160 @@
+#!/usr/bin/env bash
+
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+source "$(dirname "$0")"/common.sh
+source "$(dirname "$0")"/common_ha.sh
+
+TEST_PROGRAM_JAR_NAME=DataStreamAllroundTestProgram.jar
+TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-datastream-allround-test/target/${TEST_PROGRAM_JAR_NAME}
+FLINK_LIB_DIR=${FLINK_DIR}/lib
+JOB_ID=""
+
+function ha_cleanup() {
+  # don't call ourselves again for another signal interruption
+  trap "exit -1" INT
+  # don't call ourselves again for normal exit
+  trap "" EXIT
+
+  stop_watchdogs
+  kill_all 'StandaloneJobClusterEntryPoint'
+  shutdown_all
+  rm ${FLINK_LIB_DIR}/${TEST_PROGRAM_JAR_NAME}
+}
+
+trap ha_cleanup INT
+trap ha_cleanup EXIT
+
+function run_job() {
+local PARALLELISM=$1
+local BACKEND=$2
+local ASYNC=$3
+local INCREM=$4
+
+local CHECKPOINT_DIR="${TEST_DATA_DIR}/checkpoints/"
+
+${FLINK_DIR}/bin/standalone-job.sh start \
+--job-classname 
org.apache.flink.streaming.tests.DataStreamAllroundTestProgram \
+-p ${PARALLELISM} \
+--environment.parallelism ${PARALLELISM} \
+--test.semantics exactly-once \
+--test.simulate_failure true \
+--test.simulate_failure.num_records 200 \
+--test.simulate_failure.num_checkpoints 1 \
+--test.simulate_failure.max_failures 20 \
+--state_backend ${BACKEND} \
+--state_backend.checkpoint_directory "file://${CHECKPOINT_DIR}" \
+--state_backend.file.async ${ASYNC} \
+--state_backend.rocks.incremental ${INCREM} \
+--sequence_generator_source.sleep_time 15 \
+--sequence_generator_source.sleep_after_elements 1
+}
+
+function verify_logs_per_job() {
+local OUTPUT=$FLINK_DIR/log/*.out
+local JM_FAILURES=$1
+local EXIT_CODE=0
+local VERIFY_CHECKPOINTS=$2
+
+# verify that we have no alerts
+if ! [ `cat ${OUTPUT} | wc -l` -eq 0 ]; then
+echo "FAILURE: Alerts found at the general purpose job."
+EXIT_CODE=1
+fi
+
+# checks that all apart from the first JM recover the failed jobgraph.
+if ! [ `grep -r --include '*standalonejob*.log' 'Found 0 checkpoints in 
ZooKeeper' "${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq 1 ]; then
+echo "FAILURE: A JM did not take over, but started new job."
+EXIT_CODE=1
+fi
+
+if ! [ `grep -r --include '*standalonejob*.log' -P 'Found \d+ checkpoints 
in ZooKeeper' "${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq 
$((JM_FAILURES + 1)) ]; then
+echo "FAILURE: A JM did not take over."
+EXIT_CODE=1
+fi
+
+if [ "$VERIFY_CHECKPOINTS" = true ]; then
+# search the logs for JMs that log completed checkpoints
+if ! [ `grep -r --include '*standalonejob*.log' 'Completed checkpoint' 
"${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq $((JM_FAILURES + 1)) 
]; then
+echo "FAILURE: A JM did not execute the job."
+EXIT_CODE=1
+fi
+fi
+
+if [[ $EXIT_CODE != 0 ]]; then
+echo "One or more tests FAILED."
+exit $EXIT_CODE
+fi
+}
 
 Review comment:
   This looks like a duplicate of `common_ha.sh#verify_logs`.


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 

[jira] [Commented] (FLINK-10632) Run general purpose test job with failures in per-job mode

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10632:


tillrohrmann commented on a change in pull request #6943: [FLINK-10632][e2e] 
Running general purpose testing job with failure in per-job mode
URL: https://github.com/apache/flink/pull/6943#discussion_r229376944
 
 

 ##
 File path: 
flink-end-to-end-tests/test-scripts/test_ha_per_job_cluster_datastream.sh
 ##
 @@ -0,0 +1,160 @@
+#!/usr/bin/env bash
+
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+source "$(dirname "$0")"/common.sh
+source "$(dirname "$0")"/common_ha.sh
+
+TEST_PROGRAM_JAR_NAME=DataStreamAllroundTestProgram.jar
+TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-datastream-allround-test/target/${TEST_PROGRAM_JAR_NAME}
+FLINK_LIB_DIR=${FLINK_DIR}/lib
+JOB_ID=""
+
+function ha_cleanup() {
+  # don't call ourselves again for another signal interruption
+  trap "exit -1" INT
+  # don't call ourselves again for normal exit
+  trap "" EXIT
+
+  stop_watchdogs
+  kill_all 'StandaloneJobClusterEntryPoint'
+  shutdown_all
+  rm ${FLINK_LIB_DIR}/${TEST_PROGRAM_JAR_NAME}
+}
+
+trap ha_cleanup INT
+trap ha_cleanup EXIT
+
+function run_job() {
+local PARALLELISM=$1
+local BACKEND=$2
+local ASYNC=$3
+local INCREM=$4
+
+local CHECKPOINT_DIR="${TEST_DATA_DIR}/checkpoints/"
+
+${FLINK_DIR}/bin/standalone-job.sh start \
+--job-classname 
org.apache.flink.streaming.tests.DataStreamAllroundTestProgram \
+-p ${PARALLELISM} \
+--environment.parallelism ${PARALLELISM} \
+--test.semantics exactly-once \
+--test.simulate_failure true \
+--test.simulate_failure.num_records 200 \
+--test.simulate_failure.num_checkpoints 1 \
+--test.simulate_failure.max_failures 20 \
+--state_backend ${BACKEND} \
+--state_backend.checkpoint_directory "file://${CHECKPOINT_DIR}" \
+--state_backend.file.async ${ASYNC} \
+--state_backend.rocks.incremental ${INCREM} \
+--sequence_generator_source.sleep_time 15 \
+--sequence_generator_source.sleep_after_elements 1
+}
+
+function verify_logs_per_job() {
+local OUTPUT=$FLINK_DIR/log/*.out
+local JM_FAILURES=$1
+local EXIT_CODE=0
+local VERIFY_CHECKPOINTS=$2
+
+# verify that we have no alerts
+if ! [ `cat ${OUTPUT} | wc -l` -eq 0 ]; then
+echo "FAILURE: Alerts found at the general purpose job."
+EXIT_CODE=1
+fi
+
+# checks that all apart from the first JM recover the failed jobgraph.
+if ! [ `grep -r --include '*standalonejob*.log' 'Found 0 checkpoints in 
ZooKeeper' "${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq 1 ]; then
+echo "FAILURE: A JM did not take over, but started new job."
+EXIT_CODE=1
+fi
+
+if ! [ `grep -r --include '*standalonejob*.log' -P 'Found \d+ checkpoints 
in ZooKeeper' "${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq 
$((JM_FAILURES + 1)) ]; then
 
 Review comment:
   The standard `grep` application on Mac OS does not support `-P` because it's 
FreeBSD's grep version. On Mac OS you can pass the `-E` option to enable 
extended regex support which would support `\d+`. I think we should make it 
Unix compatible.


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


> Run general purpose test job with failures in per-job mode
> --
>
> Key: FLINK-10632
> URL: https://issues.apache.org/jira/browse/FLINK-10632
> Project: Flink
>  Issue Type: Sub-task
>  Components: E2E Tests
>Affects Versions: 

[jira] [Commented] (FLINK-10632) Run general purpose test job with failures in per-job mode

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10632:


tillrohrmann commented on a change in pull request #6943: [FLINK-10632][e2e] 
Running general purpose testing job with failure in per-job mode
URL: https://github.com/apache/flink/pull/6943#discussion_r229343747
 
 

 ##
 File path: flink-end-to-end-tests/test-scripts/test_ha_dataset.sh
 ##
 @@ -22,6 +22,20 @@ source "$(dirname "$0")"/common_ha.sh
 
 
TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-dataset-allround-test/target/DataSetAllroundTestProgram.jar
 
+function ha_cleanup() {
+  # don't call ourselves again for another signal interruption
+  trap "exit -1" INT
+  # don't call ourselves again for normal exit
+  trap "" EXIT
+
+  # kill the cluster and zookeeper
+  stop_watchdogs
+  shutdown_all
 
 Review comment:
   I think `shutdown_all` will be called by the test runner in 
`test-runner-common.sh`


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


> Run general purpose test job with failures in per-job mode
> --
>
> Key: FLINK-10632
> URL: https://issues.apache.org/jira/browse/FLINK-10632
> Project: Flink
>  Issue Type: Sub-task
>  Components: E2E Tests
>Affects Versions: 1.7.0
>Reporter: Till Rohrmann
>Assignee: Dawid Wysakowicz
>Priority: Major
>  Labels: pull-request-available
> Fix For: 1.7.0
>
>
> Similar to FLINK-8973, we should add an end-to-end which runs the general 
> datastream job with failures on a per-job cluster with HA enabled (either 
> directly the {{StandaloneJobClusterEntrypoint}} or a docker image based on 
> this entrypoint).
> We should kill the TMs as well as the cluster entrypoint and verify that the 
> job recovers.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-10632) Run general purpose test job with failures in per-job mode

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10632:


tillrohrmann commented on a change in pull request #6943: [FLINK-10632][e2e] 
Running general purpose testing job with failure in per-job mode
URL: https://github.com/apache/flink/pull/6943#discussion_r229353357
 
 

 ##
 File path: 
flink-end-to-end-tests/test-scripts/test_ha_per_job_cluster_datastream.sh
 ##
 @@ -0,0 +1,160 @@
+#!/usr/bin/env bash
+
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+source "$(dirname "$0")"/common.sh
+source "$(dirname "$0")"/common_ha.sh
+
+TEST_PROGRAM_JAR_NAME=DataStreamAllroundTestProgram.jar
+TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-datastream-allround-test/target/${TEST_PROGRAM_JAR_NAME}
+FLINK_LIB_DIR=${FLINK_DIR}/lib
+JOB_ID=""
+
+function ha_cleanup() {
+  # don't call ourselves again for another signal interruption
+  trap "exit -1" INT
+  # don't call ourselves again for normal exit
+  trap "" EXIT
+
+  stop_watchdogs
+  kill_all 'StandaloneJobClusterEntryPoint'
+  shutdown_all
+  rm ${FLINK_LIB_DIR}/${TEST_PROGRAM_JAR_NAME}
+}
+
+trap ha_cleanup INT
+trap ha_cleanup EXIT
+
+function run_job() {
+local PARALLELISM=$1
+local BACKEND=$2
+local ASYNC=$3
+local INCREM=$4
+
+local CHECKPOINT_DIR="${TEST_DATA_DIR}/checkpoints/"
+
+${FLINK_DIR}/bin/standalone-job.sh start \
+--job-classname 
org.apache.flink.streaming.tests.DataStreamAllroundTestProgram \
+-p ${PARALLELISM} \
+--environment.parallelism ${PARALLELISM} \
+--test.semantics exactly-once \
+--test.simulate_failure true \
+--test.simulate_failure.num_records 200 \
+--test.simulate_failure.num_checkpoints 1 \
+--test.simulate_failure.max_failures 20 \
+--state_backend ${BACKEND} \
+--state_backend.checkpoint_directory "file://${CHECKPOINT_DIR}" \
+--state_backend.file.async ${ASYNC} \
+--state_backend.rocks.incremental ${INCREM} \
+--sequence_generator_source.sleep_time 15 \
+--sequence_generator_source.sleep_after_elements 1
+}
+
+function verify_logs_per_job() {
+local OUTPUT=$FLINK_DIR/log/*.out
+local JM_FAILURES=$1
+local EXIT_CODE=0
+local VERIFY_CHECKPOINTS=$2
+
+# verify that we have no alerts
+if ! [ `cat ${OUTPUT} | wc -l` -eq 0 ]; then
+echo "FAILURE: Alerts found at the general purpose job."
+EXIT_CODE=1
+fi
+
+# checks that all apart from the first JM recover the failed jobgraph.
+if ! [ `grep -r --include '*standalonejob*.log' 'Found 0 checkpoints in 
ZooKeeper' "${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq 1 ]; then
+echo "FAILURE: A JM did not take over, but started new job."
+EXIT_CODE=1
+fi
+
+if ! [ `grep -r --include '*standalonejob*.log' -P 'Found \d+ checkpoints 
in ZooKeeper' "${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq 
$((JM_FAILURES + 1)) ]; then
+echo "FAILURE: A JM did not take over."
+EXIT_CODE=1
+fi
+
+if [ "$VERIFY_CHECKPOINTS" = true ]; then
+# search the logs for JMs that log completed checkpoints
+if ! [ `grep -r --include '*standalonejob*.log' 'Completed checkpoint' 
"${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq $((JM_FAILURES + 1)) 
]; then
+echo "FAILURE: A JM did not execute the job."
+EXIT_CODE=1
+fi
+fi
+
+if [[ $EXIT_CODE != 0 ]]; then
+echo "One or more tests FAILED."
+exit $EXIT_CODE
+fi
+}
+
+function run_ha_test() {
+local PARALLELISM=$1
+local BACKEND=$2
+local ASYNC=$3
+local INCREM=$4
+
+local JM_KILLS=3
+
+
+CLEARED=0
+
+# add job jar to cluster classpath
+cp ${TEST_PROGRAM_JAR} ${FLINK_LIB_DIR}
+
+# start the cluster on HA mode
+create_ha_config
+start_local_zk
+
+echo "Running on HA mode: 

[jira] [Commented] (FLINK-10632) Run general purpose test job with failures in per-job mode

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10632:


tillrohrmann commented on a change in pull request #6943: [FLINK-10632][e2e] 
Running general purpose testing job with failure in per-job mode
URL: https://github.com/apache/flink/pull/6943#discussion_r229344187
 
 

 ##
 File path: 
flink-end-to-end-tests/test-scripts/test_ha_per_job_cluster_datastream.sh
 ##
 @@ -0,0 +1,160 @@
+#!/usr/bin/env bash
+
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+source "$(dirname "$0")"/common.sh
+source "$(dirname "$0")"/common_ha.sh
+
+TEST_PROGRAM_JAR_NAME=DataStreamAllroundTestProgram.jar
+TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-datastream-allround-test/target/${TEST_PROGRAM_JAR_NAME}
+FLINK_LIB_DIR=${FLINK_DIR}/lib
+JOB_ID=""
+
+function ha_cleanup() {
+  # don't call ourselves again for another signal interruption
+  trap "exit -1" INT
+  # don't call ourselves again for normal exit
+  trap "" EXIT
+
+  stop_watchdogs
+  kill_all 'StandaloneJobClusterEntryPoint'
+  shutdown_all
 
 Review comment:
   No need to call this, since the test runner will call 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


> Run general purpose test job with failures in per-job mode
> --
>
> Key: FLINK-10632
> URL: https://issues.apache.org/jira/browse/FLINK-10632
> Project: Flink
>  Issue Type: Sub-task
>  Components: E2E Tests
>Affects Versions: 1.7.0
>Reporter: Till Rohrmann
>Assignee: Dawid Wysakowicz
>Priority: Major
>  Labels: pull-request-available
> Fix For: 1.7.0
>
>
> Similar to FLINK-8973, we should add an end-to-end which runs the general 
> datastream job with failures on a per-job cluster with HA enabled (either 
> directly the {{StandaloneJobClusterEntrypoint}} or a docker image based on 
> this entrypoint).
> We should kill the TMs as well as the cluster entrypoint and verify that the 
> job recovers.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-10632) Run general purpose test job with failures in per-job mode

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-10632:


tillrohrmann commented on a change in pull request #6943: [FLINK-10632][e2e] 
Running general purpose testing job with failure in per-job mode
URL: https://github.com/apache/flink/pull/6943#discussion_r229353456
 
 

 ##
 File path: 
flink-end-to-end-tests/test-scripts/test_ha_per_job_cluster_datastream.sh
 ##
 @@ -0,0 +1,160 @@
+#!/usr/bin/env bash
+
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+source "$(dirname "$0")"/common.sh
+source "$(dirname "$0")"/common_ha.sh
+
+TEST_PROGRAM_JAR_NAME=DataStreamAllroundTestProgram.jar
+TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-datastream-allround-test/target/${TEST_PROGRAM_JAR_NAME}
+FLINK_LIB_DIR=${FLINK_DIR}/lib
+JOB_ID=""
+
+function ha_cleanup() {
+  # don't call ourselves again for another signal interruption
+  trap "exit -1" INT
+  # don't call ourselves again for normal exit
+  trap "" EXIT
+
+  stop_watchdogs
+  kill_all 'StandaloneJobClusterEntryPoint'
+  shutdown_all
+  rm ${FLINK_LIB_DIR}/${TEST_PROGRAM_JAR_NAME}
+}
+
+trap ha_cleanup INT
+trap ha_cleanup EXIT
+
+function run_job() {
+local PARALLELISM=$1
+local BACKEND=$2
+local ASYNC=$3
+local INCREM=$4
+
+local CHECKPOINT_DIR="${TEST_DATA_DIR}/checkpoints/"
+
+${FLINK_DIR}/bin/standalone-job.sh start \
+--job-classname 
org.apache.flink.streaming.tests.DataStreamAllroundTestProgram \
+-p ${PARALLELISM} \
+--environment.parallelism ${PARALLELISM} \
+--test.semantics exactly-once \
+--test.simulate_failure true \
+--test.simulate_failure.num_records 200 \
+--test.simulate_failure.num_checkpoints 1 \
+--test.simulate_failure.max_failures 20 \
+--state_backend ${BACKEND} \
+--state_backend.checkpoint_directory "file://${CHECKPOINT_DIR}" \
+--state_backend.file.async ${ASYNC} \
+--state_backend.rocks.incremental ${INCREM} \
+--sequence_generator_source.sleep_time 15 \
+--sequence_generator_source.sleep_after_elements 1
+}
+
+function verify_logs_per_job() {
+local OUTPUT=$FLINK_DIR/log/*.out
+local JM_FAILURES=$1
+local EXIT_CODE=0
+local VERIFY_CHECKPOINTS=$2
+
+# verify that we have no alerts
+if ! [ `cat ${OUTPUT} | wc -l` -eq 0 ]; then
+echo "FAILURE: Alerts found at the general purpose job."
+EXIT_CODE=1
+fi
+
+# checks that all apart from the first JM recover the failed jobgraph.
+if ! [ `grep -r --include '*standalonejob*.log' 'Found 0 checkpoints in 
ZooKeeper' "${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq 1 ]; then
+echo "FAILURE: A JM did not take over, but started new job."
+EXIT_CODE=1
+fi
+
+if ! [ `grep -r --include '*standalonejob*.log' -P 'Found \d+ checkpoints 
in ZooKeeper' "${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq 
$((JM_FAILURES + 1)) ]; then
+echo "FAILURE: A JM did not take over."
+EXIT_CODE=1
+fi
+
+if [ "$VERIFY_CHECKPOINTS" = true ]; then
+# search the logs for JMs that log completed checkpoints
+if ! [ `grep -r --include '*standalonejob*.log' 'Completed checkpoint' 
"${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq $((JM_FAILURES + 1)) 
]; then
+echo "FAILURE: A JM did not execute the job."
+EXIT_CODE=1
+fi
+fi
+
+if [[ $EXIT_CODE != 0 ]]; then
+echo "One or more tests FAILED."
+exit $EXIT_CODE
+fi
+}
+
+function run_ha_test() {
+local PARALLELISM=$1
+local BACKEND=$2
+local ASYNC=$3
+local INCREM=$4
+
+local JM_KILLS=3
+
+
+CLEARED=0
+
+# add job jar to cluster classpath
+cp ${TEST_PROGRAM_JAR} ${FLINK_LIB_DIR}
+
+# start the cluster on HA mode
+create_ha_config
+start_local_zk
+
+echo "Running on HA mode: 

[jira] [Commented] (FLINK-10712) RestartPipelinedRegionStrategy does not restore state

[Yun Tang (JIRA)]

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

Yun Tang commented on FLINK-10712:
--

We have refactored _FailoverRegion.java_ to support fail-over with state when 
using region-failover strategy. I'll organize related code to crate a new PR in 
these days.

> RestartPipelinedRegionStrategy does not restore state
> -
>
> Key: FLINK-10712
> URL: https://issues.apache.org/jira/browse/FLINK-10712
> Project: Flink
>  Issue Type: Bug
>  Components: Distributed Coordination
>Affects Versions: 1.3.3, 1.4.2, 1.5.5, 1.6.2, 1.7.0
>Reporter: Stefan Richter
>Priority: Critical
> Fix For: 1.7.0
>
>
> RestartPipelinedRegionStrategy does not perform any state restore. This is 
> big problem because all restored regions will be restarted with empty state. 
> We need to take checkpoints into account when restoring.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[GitHub] tillrohrmann commented on a change in pull request #6943: [FLINK-10632][e2e] Running general purpose testing job with failure in per-job mode

[GitBox]

tillrohrmann commented on a change in pull request #6943: [FLINK-10632][e2e] 
Running general purpose testing job with failure in per-job mode
URL: https://github.com/apache/flink/pull/6943#discussion_r229376944
 
 

 ##
 File path: 
flink-end-to-end-tests/test-scripts/test_ha_per_job_cluster_datastream.sh
 ##
 @@ -0,0 +1,160 @@
+#!/usr/bin/env bash
+
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+source "$(dirname "$0")"/common.sh
+source "$(dirname "$0")"/common_ha.sh
+
+TEST_PROGRAM_JAR_NAME=DataStreamAllroundTestProgram.jar
+TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-datastream-allround-test/target/${TEST_PROGRAM_JAR_NAME}
+FLINK_LIB_DIR=${FLINK_DIR}/lib
+JOB_ID=""
+
+function ha_cleanup() {
+  # don't call ourselves again for another signal interruption
+  trap "exit -1" INT
+  # don't call ourselves again for normal exit
+  trap "" EXIT
+
+  stop_watchdogs
+  kill_all 'StandaloneJobClusterEntryPoint'
+  shutdown_all
+  rm ${FLINK_LIB_DIR}/${TEST_PROGRAM_JAR_NAME}
+}
+
+trap ha_cleanup INT
+trap ha_cleanup EXIT
+
+function run_job() {
+local PARALLELISM=$1
+local BACKEND=$2
+local ASYNC=$3
+local INCREM=$4
+
+local CHECKPOINT_DIR="${TEST_DATA_DIR}/checkpoints/"
+
+${FLINK_DIR}/bin/standalone-job.sh start \
+--job-classname 
org.apache.flink.streaming.tests.DataStreamAllroundTestProgram \
+-p ${PARALLELISM} \
+--environment.parallelism ${PARALLELISM} \
+--test.semantics exactly-once \
+--test.simulate_failure true \
+--test.simulate_failure.num_records 200 \
+--test.simulate_failure.num_checkpoints 1 \
+--test.simulate_failure.max_failures 20 \
+--state_backend ${BACKEND} \
+--state_backend.checkpoint_directory "file://${CHECKPOINT_DIR}" \
+--state_backend.file.async ${ASYNC} \
+--state_backend.rocks.incremental ${INCREM} \
+--sequence_generator_source.sleep_time 15 \
+--sequence_generator_source.sleep_after_elements 1
+}
+
+function verify_logs_per_job() {
+local OUTPUT=$FLINK_DIR/log/*.out
+local JM_FAILURES=$1
+local EXIT_CODE=0
+local VERIFY_CHECKPOINTS=$2
+
+# verify that we have no alerts
+if ! [ `cat ${OUTPUT} | wc -l` -eq 0 ]; then
+echo "FAILURE: Alerts found at the general purpose job."
+EXIT_CODE=1
+fi
+
+# checks that all apart from the first JM recover the failed jobgraph.
+if ! [ `grep -r --include '*standalonejob*.log' 'Found 0 checkpoints in 
ZooKeeper' "${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq 1 ]; then
+echo "FAILURE: A JM did not take over, but started new job."
+EXIT_CODE=1
+fi
+
+if ! [ `grep -r --include '*standalonejob*.log' -P 'Found \d+ checkpoints 
in ZooKeeper' "${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq 
$((JM_FAILURES + 1)) ]; then
 
 Review comment:
   The standard `grep` application on Mac OS does not support `-P` because it's 
FreeBSD's grep version. On Mac OS you can pass the `-E` option to enable 
extended regex support which would support `\d+`. I think we should make it 
Unix compatible.


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 commented on a change in pull request #6943: [FLINK-10632][e2e] Running general purpose testing job with failure in per-job mode

[GitBox]

tillrohrmann commented on a change in pull request #6943: [FLINK-10632][e2e] 
Running general purpose testing job with failure in per-job mode
URL: https://github.com/apache/flink/pull/6943#discussion_r229344187
 
 

 ##
 File path: 
flink-end-to-end-tests/test-scripts/test_ha_per_job_cluster_datastream.sh
 ##
 @@ -0,0 +1,160 @@
+#!/usr/bin/env bash
+
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+source "$(dirname "$0")"/common.sh
+source "$(dirname "$0")"/common_ha.sh
+
+TEST_PROGRAM_JAR_NAME=DataStreamAllroundTestProgram.jar
+TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-datastream-allround-test/target/${TEST_PROGRAM_JAR_NAME}
+FLINK_LIB_DIR=${FLINK_DIR}/lib
+JOB_ID=""
+
+function ha_cleanup() {
+  # don't call ourselves again for another signal interruption
+  trap "exit -1" INT
+  # don't call ourselves again for normal exit
+  trap "" EXIT
+
+  stop_watchdogs
+  kill_all 'StandaloneJobClusterEntryPoint'
+  shutdown_all
 
 Review comment:
   No need to call this, since the test runner will call 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 commented on a change in pull request #6943: [FLINK-10632][e2e] Running general purpose testing job with failure in per-job mode

[GitBox]

tillrohrmann commented on a change in pull request #6943: [FLINK-10632][e2e] 
Running general purpose testing job with failure in per-job mode
URL: https://github.com/apache/flink/pull/6943#discussion_r229353357
 
 

 ##
 File path: 
flink-end-to-end-tests/test-scripts/test_ha_per_job_cluster_datastream.sh
 ##
 @@ -0,0 +1,160 @@
+#!/usr/bin/env bash
+
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+source "$(dirname "$0")"/common.sh
+source "$(dirname "$0")"/common_ha.sh
+
+TEST_PROGRAM_JAR_NAME=DataStreamAllroundTestProgram.jar
+TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-datastream-allround-test/target/${TEST_PROGRAM_JAR_NAME}
+FLINK_LIB_DIR=${FLINK_DIR}/lib
+JOB_ID=""
+
+function ha_cleanup() {
+  # don't call ourselves again for another signal interruption
+  trap "exit -1" INT
+  # don't call ourselves again for normal exit
+  trap "" EXIT
+
+  stop_watchdogs
+  kill_all 'StandaloneJobClusterEntryPoint'
+  shutdown_all
+  rm ${FLINK_LIB_DIR}/${TEST_PROGRAM_JAR_NAME}
+}
+
+trap ha_cleanup INT
+trap ha_cleanup EXIT
+
+function run_job() {
+local PARALLELISM=$1
+local BACKEND=$2
+local ASYNC=$3
+local INCREM=$4
+
+local CHECKPOINT_DIR="${TEST_DATA_DIR}/checkpoints/"
+
+${FLINK_DIR}/bin/standalone-job.sh start \
+--job-classname 
org.apache.flink.streaming.tests.DataStreamAllroundTestProgram \
+-p ${PARALLELISM} \
+--environment.parallelism ${PARALLELISM} \
+--test.semantics exactly-once \
+--test.simulate_failure true \
+--test.simulate_failure.num_records 200 \
+--test.simulate_failure.num_checkpoints 1 \
+--test.simulate_failure.max_failures 20 \
+--state_backend ${BACKEND} \
+--state_backend.checkpoint_directory "file://${CHECKPOINT_DIR}" \
+--state_backend.file.async ${ASYNC} \
+--state_backend.rocks.incremental ${INCREM} \
+--sequence_generator_source.sleep_time 15 \
+--sequence_generator_source.sleep_after_elements 1
+}
+
+function verify_logs_per_job() {
+local OUTPUT=$FLINK_DIR/log/*.out
+local JM_FAILURES=$1
+local EXIT_CODE=0
+local VERIFY_CHECKPOINTS=$2
+
+# verify that we have no alerts
+if ! [ `cat ${OUTPUT} | wc -l` -eq 0 ]; then
+echo "FAILURE: Alerts found at the general purpose job."
+EXIT_CODE=1
+fi
+
+# checks that all apart from the first JM recover the failed jobgraph.
+if ! [ `grep -r --include '*standalonejob*.log' 'Found 0 checkpoints in 
ZooKeeper' "${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq 1 ]; then
+echo "FAILURE: A JM did not take over, but started new job."
+EXIT_CODE=1
+fi
+
+if ! [ `grep -r --include '*standalonejob*.log' -P 'Found \d+ checkpoints 
in ZooKeeper' "${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq 
$((JM_FAILURES + 1)) ]; then
+echo "FAILURE: A JM did not take over."
+EXIT_CODE=1
+fi
+
+if [ "$VERIFY_CHECKPOINTS" = true ]; then
+# search the logs for JMs that log completed checkpoints
+if ! [ `grep -r --include '*standalonejob*.log' 'Completed checkpoint' 
"${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq $((JM_FAILURES + 1)) 
]; then
+echo "FAILURE: A JM did not execute the job."
+EXIT_CODE=1
+fi
+fi
+
+if [[ $EXIT_CODE != 0 ]]; then
+echo "One or more tests FAILED."
+exit $EXIT_CODE
+fi
+}
+
+function run_ha_test() {
+local PARALLELISM=$1
+local BACKEND=$2
+local ASYNC=$3
+local INCREM=$4
+
+local JM_KILLS=3
+
+
+CLEARED=0
+
+# add job jar to cluster classpath
+cp ${TEST_PROGRAM_JAR} ${FLINK_LIB_DIR}
+
+# start the cluster on HA mode
+create_ha_config
+start_local_zk
+
+echo "Running on HA mode: parallelism=${PARALLELISM}, backend=${BACKEND}, 
asyncSnapshots=${ASYNC}, and incremSnapshots=${INCREM}."
+
+# submit a job in detached mode and let it run
+run_job ${PARALLELISM} ${BACKEND} ${ASYNC} ${INCREM}
+
+start_taskmanagers 

[GitHub] tillrohrmann commented on a change in pull request #6943: [FLINK-10632][e2e] Running general purpose testing job with failure in per-job mode

[GitBox]

tillrohrmann commented on a change in pull request #6943: [FLINK-10632][e2e] 
Running general purpose testing job with failure in per-job mode
URL: https://github.com/apache/flink/pull/6943#discussion_r229343747
 
 

 ##
 File path: flink-end-to-end-tests/test-scripts/test_ha_dataset.sh
 ##
 @@ -22,6 +22,20 @@ source "$(dirname "$0")"/common_ha.sh
 
 
TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-dataset-allround-test/target/DataSetAllroundTestProgram.jar
 
+function ha_cleanup() {
+  # don't call ourselves again for another signal interruption
+  trap "exit -1" INT
+  # don't call ourselves again for normal exit
+  trap "" EXIT
+
+  # kill the cluster and zookeeper
+  stop_watchdogs
+  shutdown_all
 
 Review comment:
   I think `shutdown_all` will be called by the test runner in 
`test-runner-common.sh`


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 commented on a change in pull request #6943: [FLINK-10632][e2e] Running general purpose testing job with failure in per-job mode

[GitBox]

tillrohrmann commented on a change in pull request #6943: [FLINK-10632][e2e] 
Running general purpose testing job with failure in per-job mode
URL: https://github.com/apache/flink/pull/6943#discussion_r229353456
 
 

 ##
 File path: 
flink-end-to-end-tests/test-scripts/test_ha_per_job_cluster_datastream.sh
 ##
 @@ -0,0 +1,160 @@
+#!/usr/bin/env bash
+
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+source "$(dirname "$0")"/common.sh
+source "$(dirname "$0")"/common_ha.sh
+
+TEST_PROGRAM_JAR_NAME=DataStreamAllroundTestProgram.jar
+TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-datastream-allround-test/target/${TEST_PROGRAM_JAR_NAME}
+FLINK_LIB_DIR=${FLINK_DIR}/lib
+JOB_ID=""
+
+function ha_cleanup() {
+  # don't call ourselves again for another signal interruption
+  trap "exit -1" INT
+  # don't call ourselves again for normal exit
+  trap "" EXIT
+
+  stop_watchdogs
+  kill_all 'StandaloneJobClusterEntryPoint'
+  shutdown_all
+  rm ${FLINK_LIB_DIR}/${TEST_PROGRAM_JAR_NAME}
+}
+
+trap ha_cleanup INT
+trap ha_cleanup EXIT
+
+function run_job() {
+local PARALLELISM=$1
+local BACKEND=$2
+local ASYNC=$3
+local INCREM=$4
+
+local CHECKPOINT_DIR="${TEST_DATA_DIR}/checkpoints/"
+
+${FLINK_DIR}/bin/standalone-job.sh start \
+--job-classname 
org.apache.flink.streaming.tests.DataStreamAllroundTestProgram \
+-p ${PARALLELISM} \
+--environment.parallelism ${PARALLELISM} \
+--test.semantics exactly-once \
+--test.simulate_failure true \
+--test.simulate_failure.num_records 200 \
+--test.simulate_failure.num_checkpoints 1 \
+--test.simulate_failure.max_failures 20 \
+--state_backend ${BACKEND} \
+--state_backend.checkpoint_directory "file://${CHECKPOINT_DIR}" \
+--state_backend.file.async ${ASYNC} \
+--state_backend.rocks.incremental ${INCREM} \
+--sequence_generator_source.sleep_time 15 \
+--sequence_generator_source.sleep_after_elements 1
+}
+
+function verify_logs_per_job() {
+local OUTPUT=$FLINK_DIR/log/*.out
+local JM_FAILURES=$1
+local EXIT_CODE=0
+local VERIFY_CHECKPOINTS=$2
+
+# verify that we have no alerts
+if ! [ `cat ${OUTPUT} | wc -l` -eq 0 ]; then
+echo "FAILURE: Alerts found at the general purpose job."
+EXIT_CODE=1
+fi
+
+# checks that all apart from the first JM recover the failed jobgraph.
+if ! [ `grep -r --include '*standalonejob*.log' 'Found 0 checkpoints in 
ZooKeeper' "${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq 1 ]; then
+echo "FAILURE: A JM did not take over, but started new job."
+EXIT_CODE=1
+fi
+
+if ! [ `grep -r --include '*standalonejob*.log' -P 'Found \d+ checkpoints 
in ZooKeeper' "${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq 
$((JM_FAILURES + 1)) ]; then
+echo "FAILURE: A JM did not take over."
+EXIT_CODE=1
+fi
+
+if [ "$VERIFY_CHECKPOINTS" = true ]; then
+# search the logs for JMs that log completed checkpoints
+if ! [ `grep -r --include '*standalonejob*.log' 'Completed checkpoint' 
"${FLINK_DIR}/log/" | cut -d ":" -f 1 | uniq | wc -l` -eq $((JM_FAILURES + 1)) 
]; then
+echo "FAILURE: A JM did not execute the job."
+EXIT_CODE=1
+fi
+fi
+
+if [[ $EXIT_CODE != 0 ]]; then
+echo "One or more tests FAILED."
+exit $EXIT_CODE
+fi
+}
+
+function run_ha_test() {
+local PARALLELISM=$1
+local BACKEND=$2
+local ASYNC=$3
+local INCREM=$4
+
+local JM_KILLS=3
+
+
+CLEARED=0
+
+# add job jar to cluster classpath
+cp ${TEST_PROGRAM_JAR} ${FLINK_LIB_DIR}
+
+# start the cluster on HA mode
+create_ha_config
+start_local_zk
+
+echo "Running on HA mode: parallelism=${PARALLELISM}, backend=${BACKEND}, 
asyncSnapshots=${ASYNC}, and incremSnapshots=${INCREM}."
+
+# submit a job in detached mode and let it run
+run_job ${PARALLELISM} ${BACKEND} ${ASYNC} ${INCREM}
+
+start_taskmanagers