Re: [DISCUSS] FLIP-381: Deprecate configuration getters/setters that return/set complex Java objects

[Junrui Lee]

Hi Hang,

Thank you for your feedback. Regarding your suggestion to delete the
private field 'storage', I want to clarify that in this FLIP, we are only
deprecating the getter and setter methods for CheckpointConfig#storage.
Therefore, we cannot directly remove the 'storage' field at this moment.
However, we will consider removing it altogether in FLINK-2.0.

Best regards, Junrui

Hang Ruan  于2023年11月3日周五 12:04写道：

> Thanks Junrui for driving the proposal.
>
> +1 from my side. This FLIP will help to make the configuration clearer for
> users.
>
> ps: We should also delete the private field `storage` as its getter and
> setter are deleted and it is marked as `@Deprecated`. This is not written
> in the FLIP.
>
> Best,
> Hang
>
> Yuxin Tan  于2023年11月3日周五 11:30写道：
>
> > Thanks Junrui for driving the proposal.
> >
> > +1 for this proposal. I believe this change will enhance the usability of
> > Flink configuration for both users and developers, while also ensuring
> > consistency across various types of configurations.
> >
> > Best,
> > Yuxin
> >
> >
> > Lijie Wang  于2023年11月3日周五 10:59写道：
> >
> > > Thanks Junrui for driving this.
> > >
> > > Making configurations simple and consistent has great benefits for both
> > > users and devs. +1 for the proposal.
> > >
> > > Best,
> > > Lijie
> > >
> > > weijie guo  于2023年11月2日周四 16:49写道：
> > >
> > > > Thanks Junrui for driving this proposal!
> > > >
> > > > I believe this is helpful for the new Process Function API. Because
> we
> > > > don't need to move some related class/components from flink-core to a
> > > pure
> > > > API module (maybe, called flink-core-api) after this. Even though the
> > > FLIP
> > > > related to new API is in preparation atm, I still want to emphasize
> our
> > > > goal is that user application should no longer depend on these stuff.
> > So
> > > > I'm + 1 for this proposal.
> > > >
> > > >
> > > > Best regards,
> > > >
> > > > Weijie
> > > >
> > > >
> > > > Zhu Zhu  于2023年11月2日周四 16:00写道：
> > > >
> > > > > Thanks Junrui for creating the FLIP and kicking off this
> discussion.
> > > > >
> > > > > The community has been constantly striving to unify and simplify
> the
> > > > > configuration layer of Flink. Some progress has already been made,
> > > > > such as FLINK-29379. However, the compatibility of public
> interfaces
> > > > > poses an obstacle to completing the task. The release of Flink 2.0
> > > > > presents a great opportunity to accomplish this goal.
> > > > >
> > > > > +1 for the proposal.
> > > > >
> > > > > Thanks,
> > > > > Zhu
> > > > >
> > > > > Rui Fan <1996fan...@gmail.com> 于2023年11月2日周四 10:27写道：
> > > > >
> > > > > > Thanks Junrui for driving this proposal!
> > > > > >
> > > > > > ConfigOption is easy to use for flink users, easy to manage
> options
> > > > > > for flink platform maintainers, and easy to maintain for flink
> > > > developers
> > > > > > and flink community.
> > > > > >
> > > > > > So big +1 for this proposal!
> > > > > >
> > > > > > Best,
> > > > > > Rui
> > > > > >
> > > > > > On Thu, Nov 2, 2023 at 10:10 AM Junrui Lee 
> > > > wrote:
> > > > > >
> > > > > > > Hi devs,
> > > > > > >
> > > > > > > I would like to start a discussion on FLIP-381: Deprecate
> > > > configuration
> > > > > > > getters/setters that return/set complex Java objects[1].
> > > > > > >
> > > > > > > Currently, the job configuration in FLINK is spread out across
> > > > > different
> > > > > > > components, which leads to inconsistencies and confusion. To
> > > address
> > > > > this
> > > > > > > issue, it is necessary to migrate non-ConfigOption complex Java
> > > > objects
> > > > > > to
> > > > > > > use ConfigOption and adopt a single Configuration object to
> host
> > > all
> > > > > the
> > > > > > > configuration.
> > > > > > > However, there is a significant blocker in implementing this
> > > > solution.
> > > > > > > These complex Java objects in StreamExecutionEnvironment,
> > > > > > CheckpointConfig,
> > > > > > > and ExecutionConfig have already been exposed through the
> public
> > > API,
> > > > > > > making it challenging to modify the existing implementation.
> > > > > > >
> > > > > > > Therefore, I propose to deprecate these Java objects and their
> > > > > > > corresponding getter/setter interfaces, ultimately removing
> them
> > in
> > > > > > > FLINK-2.0.
> > > > > > >
> > > > > > > Your feedback and thoughts on this proposal are highly
> > appreciated.
> > > > > > >
> > > > > > > Best regards,
> > > > > > > Junrui Lee
> > > > > > >
> > > > > > > [1]
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=278464992
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
>

Re: [DISCUSS] FLIP-381: Deprecate configuration getters/setters that return/set complex Java objects

[Hang Ruan]

Thanks Junrui for driving the proposal.

+1 from my side. This FLIP will help to make the configuration clearer for
users.

ps: We should also delete the private field `storage` as its getter and
setter are deleted and it is marked as `@Deprecated`. This is not written
in the FLIP.

Best,
Hang

Yuxin Tan  于2023年11月3日周五 11:30写道：

> Thanks Junrui for driving the proposal.
>
> +1 for this proposal. I believe this change will enhance the usability of
> Flink configuration for both users and developers, while also ensuring
> consistency across various types of configurations.
>
> Best,
> Yuxin
>
>
> Lijie Wang  于2023年11月3日周五 10:59写道：
>
> > Thanks Junrui for driving this.
> >
> > Making configurations simple and consistent has great benefits for both
> > users and devs. +1 for the proposal.
> >
> > Best,
> > Lijie
> >
> > weijie guo  于2023年11月2日周四 16:49写道：
> >
> > > Thanks Junrui for driving this proposal!
> > >
> > > I believe this is helpful for the new Process Function API. Because we
> > > don't need to move some related class/components from flink-core to a
> > pure
> > > API module (maybe, called flink-core-api) after this. Even though the
> > FLIP
> > > related to new API is in preparation atm, I still want to emphasize our
> > > goal is that user application should no longer depend on these stuff.
> So
> > > I'm + 1 for this proposal.
> > >
> > >
> > > Best regards,
> > >
> > > Weijie
> > >
> > >
> > > Zhu Zhu  于2023年11月2日周四 16:00写道：
> > >
> > > > Thanks Junrui for creating the FLIP and kicking off this discussion.
> > > >
> > > > The community has been constantly striving to unify and simplify the
> > > > configuration layer of Flink. Some progress has already been made,
> > > > such as FLINK-29379. However, the compatibility of public interfaces
> > > > poses an obstacle to completing the task. The release of Flink 2.0
> > > > presents a great opportunity to accomplish this goal.
> > > >
> > > > +1 for the proposal.
> > > >
> > > > Thanks,
> > > > Zhu
> > > >
> > > > Rui Fan <1996fan...@gmail.com> 于2023年11月2日周四 10:27写道：
> > > >
> > > > > Thanks Junrui for driving this proposal!
> > > > >
> > > > > ConfigOption is easy to use for flink users, easy to manage options
> > > > > for flink platform maintainers, and easy to maintain for flink
> > > developers
> > > > > and flink community.
> > > > >
> > > > > So big +1 for this proposal!
> > > > >
> > > > > Best,
> > > > > Rui
> > > > >
> > > > > On Thu, Nov 2, 2023 at 10:10 AM Junrui Lee 
> > > wrote:
> > > > >
> > > > > > Hi devs,
> > > > > >
> > > > > > I would like to start a discussion on FLIP-381: Deprecate
> > > configuration
> > > > > > getters/setters that return/set complex Java objects[1].
> > > > > >
> > > > > > Currently, the job configuration in FLINK is spread out across
> > > > different
> > > > > > components, which leads to inconsistencies and confusion. To
> > address
> > > > this
> > > > > > issue, it is necessary to migrate non-ConfigOption complex Java
> > > objects
> > > > > to
> > > > > > use ConfigOption and adopt a single Configuration object to host
> > all
> > > > the
> > > > > > configuration.
> > > > > > However, there is a significant blocker in implementing this
> > > solution.
> > > > > > These complex Java objects in StreamExecutionEnvironment,
> > > > > CheckpointConfig,
> > > > > > and ExecutionConfig have already been exposed through the public
> > API,
> > > > > > making it challenging to modify the existing implementation.
> > > > > >
> > > > > > Therefore, I propose to deprecate these Java objects and their
> > > > > > corresponding getter/setter interfaces, ultimately removing them
> in
> > > > > > FLINK-2.0.
> > > > > >
> > > > > > Your feedback and thoughts on this proposal are highly
> appreciated.
> > > > > >
> > > > > > Best regards,
> > > > > > Junrui Lee
> > > > > >
> > > > > > [1]
> > > > > >
> > > > >
> > > >
> > >
> >
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=278464992
> > > > > >
> > > > >
> > > >
> > >
> >
>

[jira] [Created] (FLINK-33446) SubQueryAntiJoinTest#testMultiNotExistsWithCorrelatedOnWhere_NestedCorrelation doesn't produce the correct plan

[Shengkai Fang (Jira)]

Shengkai Fang created FLINK-33446:
-

 Summary: 
SubQueryAntiJoinTest#testMultiNotExistsWithCorrelatedOnWhere_NestedCorrelation 
doesn't produce the correct plan
 Key: FLINK-33446
 URL: https://issues.apache.org/jira/browse/FLINK-33446
 Project: Flink
  Issue Type: Bug
  Components: Table SQL / Planner
Affects Versions: 1.17.2, 1.19.0, 1.18.1
Reporter: Shengkai Fang


Although this test doesn't throw an exception, you can find the final produce 3 
columns rather than 2 columns after optimization.

{code:java}
LogicalProject(inputs=[0..1], exprs=[[$4]])
+- LogicalFilter(condition=[IS NULL($5)])
   +- LogicalJoin(condition=[AND(=($0, $2), =($1, $3))], joinType=[left])
  :- LogicalProject(exprs=[[+(2, $0), +(3, $1)]])
  :  +- LogicalTableScan(table=[[default_catalog, default_database, l, 
source: [TestTableSource(a, b, c)]]])
  +- LogicalProject(inputs=[0..2], exprs=[[true]])
 +- LogicalAggregate(group=[{0, 1, 2}])
+- LogicalProject(inputs=[0..2])
   +- LogicalFilter(condition=[IS NULL($3)])
  +- LogicalJoin(condition=[true], joinType=[left])
 :- LogicalFilter(condition=[IS NOT NULL($0)])
 :  +- LogicalProject(exprs=[[+($0, 1)]])
 : +- LogicalTableScan(table=[[default_catalog, 
default_database, r, source: [TestTableSource(d, e, f)]]])
 +- LogicalProject(inputs=[0..1], exprs=[[true]])
+- LogicalAggregate(group=[{0, 1}])
   +- LogicalProject(exprs=[[$3, $0]])
  +- LogicalFilter(condition=[AND(=($1, $0), 
=(CAST($2):BIGINT, $3))])
 +- LogicalProject(exprs=[[+($0, 4), +($0, 5), 
+($0, 6), CAST(+($0, 6)):BIGINT]])
+- 
LogicalTableScan(table=[[default_catalog, default_database, t, source: 
[TestTableSource(i, j, k)]]])

{code}

After digging, I think it's the SubQueryRemoveRule doesn't generate the 
Correlate but generates the Join node, which causes the failure of the 
decorrelation. For a quick fix, I think we should throw an exception to notify 
users it's not a supported feature in the Flink. 

There might exist 2 ways to fix this issue:
1. Expand subquery when converting SQL to rel.  After experimenting with 
calcite, I find the Sql2RelConverter generates the correct plan.

{code:java}
LogicalProject(inputs=[0..1])
+- LogicalFilter(condition=[IS NULL($2)])
   +- LogicalCorrelate(correlation=[$cor7], joinType=[left], 
requiredColumns=[{0, 1}])
  :- LogicalProject(exprs=[[+(2, $0), +(3, $1)]])
  :  +- LogicalTableScan(table=[[default_catalog, default_database, l, 
source: [TestTableSource(a, b, c)]]])
  +- LogicalAggregate(group=[{}], agg#0=[MIN($0)])
 +- LogicalProject(exprs=[[true]])
+- LogicalFilter(condition=[AND(=($0, $cor7.d2), IS NULL($1))])
   +- LogicalCorrelate(correlation=[$cor4], joinType=[left], 
requiredColumns=[{0}])
  :- LogicalProject(inputs=[0])
  :  +- LogicalTableScan(table=[[default_catalog, 
default_database, r, source: [TestTableSource(d1, e, f)]]])
  +- LogicalAggregate(group=[{}], agg#0=[MIN($0)])
 +- LogicalProject(exprs=[[true]])
+- LogicalFilter(condition=[AND(=($0, $cor4.d1), =($1, 
$cor4.d1), =(CAST($2):BIGINT, $cor7.d3))])
   +- LogicalProject(exprs=[[+($0, 4), +($0, 5), +($0, 
6)]])
  +- LogicalTableScan(table=[[default_catalog, 
default_database, t, source: [TestTableSource(i, j, k)]]])
{code}

You can find the new plan uses a correlate node rather than join node.

2. CALCITE-4686 might fix this problem by removing the nested correlation node.









--
This message was sent by Atlassian Jira
(v8.20.10#820010)

Re: [DISCUSS] FLIP-381: Deprecate configuration getters/setters that return/set complex Java objects

[Yuxin Tan]

Thanks Junrui for driving the proposal.

+1 for this proposal. I believe this change will enhance the usability of
Flink configuration for both users and developers, while also ensuring
consistency across various types of configurations.

Best,
Yuxin


Lijie Wang  于2023年11月3日周五 10:59写道：

> Thanks Junrui for driving this.
>
> Making configurations simple and consistent has great benefits for both
> users and devs. +1 for the proposal.
>
> Best,
> Lijie
>
> weijie guo  于2023年11月2日周四 16:49写道：
>
> > Thanks Junrui for driving this proposal!
> >
> > I believe this is helpful for the new Process Function API. Because we
> > don't need to move some related class/components from flink-core to a
> pure
> > API module (maybe, called flink-core-api) after this. Even though the
> FLIP
> > related to new API is in preparation atm, I still want to emphasize our
> > goal is that user application should no longer depend on these stuff. So
> > I'm + 1 for this proposal.
> >
> >
> > Best regards,
> >
> > Weijie
> >
> >
> > Zhu Zhu  于2023年11月2日周四 16:00写道：
> >
> > > Thanks Junrui for creating the FLIP and kicking off this discussion.
> > >
> > > The community has been constantly striving to unify and simplify the
> > > configuration layer of Flink. Some progress has already been made,
> > > such as FLINK-29379. However, the compatibility of public interfaces
> > > poses an obstacle to completing the task. The release of Flink 2.0
> > > presents a great opportunity to accomplish this goal.
> > >
> > > +1 for the proposal.
> > >
> > > Thanks,
> > > Zhu
> > >
> > > Rui Fan <1996fan...@gmail.com> 于2023年11月2日周四 10:27写道：
> > >
> > > > Thanks Junrui for driving this proposal!
> > > >
> > > > ConfigOption is easy to use for flink users, easy to manage options
> > > > for flink platform maintainers, and easy to maintain for flink
> > developers
> > > > and flink community.
> > > >
> > > > So big +1 for this proposal!
> > > >
> > > > Best,
> > > > Rui
> > > >
> > > > On Thu, Nov 2, 2023 at 10:10 AM Junrui Lee 
> > wrote:
> > > >
> > > > > Hi devs,
> > > > >
> > > > > I would like to start a discussion on FLIP-381: Deprecate
> > configuration
> > > > > getters/setters that return/set complex Java objects[1].
> > > > >
> > > > > Currently, the job configuration in FLINK is spread out across
> > > different
> > > > > components, which leads to inconsistencies and confusion. To
> address
> > > this
> > > > > issue, it is necessary to migrate non-ConfigOption complex Java
> > objects
> > > > to
> > > > > use ConfigOption and adopt a single Configuration object to host
> all
> > > the
> > > > > configuration.
> > > > > However, there is a significant blocker in implementing this
> > solution.
> > > > > These complex Java objects in StreamExecutionEnvironment,
> > > > CheckpointConfig,
> > > > > and ExecutionConfig have already been exposed through the public
> API,
> > > > > making it challenging to modify the existing implementation.
> > > > >
> > > > > Therefore, I propose to deprecate these Java objects and their
> > > > > corresponding getter/setter interfaces, ultimately removing them in
> > > > > FLINK-2.0.
> > > > >
> > > > > Your feedback and thoughts on this proposal are highly appreciated.
> > > > >
> > > > > Best regards,
> > > > > Junrui Lee
> > > > >
> > > > > [1]
> > > > >
> > > >
> > >
> >
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=278464992
> > > > >
> > > >
> > >
> >
>

Re: [DISCUSS] FLIP-381: Deprecate configuration getters/setters that return/set complex Java objects

[Lijie Wang]

Thanks Junrui for driving this.

Making configurations simple and consistent has great benefits for both
users and devs. +1 for the proposal.

Best,
Lijie

weijie guo  于2023年11月2日周四 16:49写道：

> Thanks Junrui for driving this proposal!
>
> I believe this is helpful for the new Process Function API. Because we
> don't need to move some related class/components from flink-core to a pure
> API module (maybe, called flink-core-api) after this. Even though the FLIP
> related to new API is in preparation atm, I still want to emphasize our
> goal is that user application should no longer depend on these stuff. So
> I'm + 1 for this proposal.
>
>
> Best regards,
>
> Weijie
>
>
> Zhu Zhu  于2023年11月2日周四 16:00写道：
>
> > Thanks Junrui for creating the FLIP and kicking off this discussion.
> >
> > The community has been constantly striving to unify and simplify the
> > configuration layer of Flink. Some progress has already been made,
> > such as FLINK-29379. However, the compatibility of public interfaces
> > poses an obstacle to completing the task. The release of Flink 2.0
> > presents a great opportunity to accomplish this goal.
> >
> > +1 for the proposal.
> >
> > Thanks,
> > Zhu
> >
> > Rui Fan <1996fan...@gmail.com> 于2023年11月2日周四 10:27写道：
> >
> > > Thanks Junrui for driving this proposal!
> > >
> > > ConfigOption is easy to use for flink users, easy to manage options
> > > for flink platform maintainers, and easy to maintain for flink
> developers
> > > and flink community.
> > >
> > > So big +1 for this proposal!
> > >
> > > Best,
> > > Rui
> > >
> > > On Thu, Nov 2, 2023 at 10:10 AM Junrui Lee 
> wrote:
> > >
> > > > Hi devs,
> > > >
> > > > I would like to start a discussion on FLIP-381: Deprecate
> configuration
> > > > getters/setters that return/set complex Java objects[1].
> > > >
> > > > Currently, the job configuration in FLINK is spread out across
> > different
> > > > components, which leads to inconsistencies and confusion. To address
> > this
> > > > issue, it is necessary to migrate non-ConfigOption complex Java
> objects
> > > to
> > > > use ConfigOption and adopt a single Configuration object to host all
> > the
> > > > configuration.
> > > > However, there is a significant blocker in implementing this
> solution.
> > > > These complex Java objects in StreamExecutionEnvironment,
> > > CheckpointConfig,
> > > > and ExecutionConfig have already been exposed through the public API,
> > > > making it challenging to modify the existing implementation.
> > > >
> > > > Therefore, I propose to deprecate these Java objects and their
> > > > corresponding getter/setter interfaces, ultimately removing them in
> > > > FLINK-2.0.
> > > >
> > > > Your feedback and thoughts on this proposal are highly appreciated.
> > > >
> > > > Best regards,
> > > > Junrui Lee
> > > >
> > > > [1]
> > > >
> > >
> >
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=278464992
> > > >
> > >
> >
>

[jira] [Created] (FLINK-33445) Translate DataSet migration guideline to Chinese

[Wencong Liu (Jira)]

Wencong Liu created FLINK-33445:
---

 Summary: Translate DataSet migration guideline to Chinese
 Key: FLINK-33445
 URL: https://issues.apache.org/jira/browse/FLINK-33445
 Project: Flink
  Issue Type: Improvement
  Components: Documentation
Affects Versions: 1.19.0
Reporter: Wencong Liu
 Fix For: 1.19.0


The [FLIINK-33041|https://issues.apache.org/jira/browse/FLINK-33041] about 
adding an introduction about how to migrate DataSet API to DataStream has been 
merged into master branch. Here is the link in the Flink website: [How to 
Migrate from DataSet to DataStream | Apache 
Flink|https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/dataset_migration/]

According to the [contribution 
guidelines|https://flink.apache.org/how-to-contribute/contribute-documentation/#chinese-documentation-translation],
 we should add an identical markdown file in {{content.zh/}} and translate it 
to Chinese. Any community volunteers are welcomed to take this task.

 



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[jira] [Created] (FLINK-33444) Fixing the stability of "org.apache.flink.connectors.hive.TableEnvHiveConnectorITCase.testUDTF"

[Krishna Anandan Ganesan (Jira)]

Krishna Anandan Ganesan created FLINK-33444:
---

 Summary: Fixing the stability of 
"org.apache.flink.connectors.hive.TableEnvHiveConnectorITCase.testUDTF"
 Key: FLINK-33444
 URL: https://issues.apache.org/jira/browse/FLINK-33444
 Project: Flink
  Issue Type: Bug
  Components: Runtime / State Backends
Reporter: Krishna Anandan Ganesan


We are proposing a solution to fix potential flakiness in the following test:
{code:java}
org.apache.flink.connectors.hive.TableEnvHiveConnectorITCase#testUDTF {code}
*STEPS TO REPRODUCE THE ISSUE:*
 * The following command can be run to execute the test with the 
[NonDex|https://github.com/TestingResearchIllinois/NonDex] plugin:

{code:java}
mvn -pl flink-connectors/flink-connector-hive 
edu.illinois:nondex-maven-plugin:2.1.1:nondex 
-Dtest=org.apache.flink.connectors.hive.TableEnvHiveConnectorITCase#testUDTF
{code}

 * The following error will be encountered:

{code:java}
[ERROR] Failures: 
[ERROR]   TableEnvHiveConnectorITCase.testUDTF:270->lambda$testUDTF$5:305 
expected: "[+I[{1=a, 2=b}], +I[{3=c}]]"
 but was: "[+I[{2=b, 1=a}], +I[{3=c}]]"
[INFO] 
[ERROR] Tests run: 1, Failures: 1, Errors: 0, Skipped
{code}

*ROOT CAUSE ANALYSIS:*
The test is currently flaky because of the assumption that the order of 
elements received in the _results_variable will be consistent. There are 
currently two versions of query output that can be stored in _results_:

# The actual order that is expected where the output of the map attribute is 
{1=a,2=b}.
# The other order which is shown in the error extract above where the ordering 
of the map attribute changes to {2=b,1=a}.

*POTENTIAL FIX:*
* The fix that I can suggest/have ready to raise a PR for is introducing 
another assertion on the second variant of the query output.
* By asserting whether the contents in _results_ are in one of the two orders, 
we can ascertain that the expected attributes with their contents are received 
as expected should the order in which they are received, not matter.

Please share your thoughts on this finding and let me know if any other 
potential fix is possible for this test.

 



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

Re: Heartbeat of TaskManager with id container

[Y SREEKARA BHARGAVA REDDY]

Hi Xiangyu,

I have one issue,

I am using *Flink** 1.16* version, How can i specify log4j.properties for
the flink run command line along with my job.
every job i need to pass different log file.

looks like below one is not working: -Dlog4j.configurationFile=
Please help me with correct config for the flink run command.

Regards,
Nagireddy Y.




On Thu, Aug 3, 2023 at 7:24 AM xiangyu feng  wrote:

> Hi ynagireddy4u,
>
> We have met this exception before. Usually it is caused by following
> reasons:
>
> 1), TaskManager is too busy with other works to send the heartbeat to
> JobMaster or TaskManager process might already exited;
> 2), There might be a network issues between this TaskManager and JobMaster;
> 3), In certain cases, JobMaster actor might also being too busy to process
> the RPC requests from TaskManager;
>
> Pls check if your problem fits the above situations.
>
> Best,
> Xiangyu
>
>
> Y SREEKARA BHARGAVA REDDY  于2023年7月31日周一 20:49写道：
>
>> Hi Team,
>>
>> Did any one face the below exception.
>> If yes, please share the resolution.
>>
>>
>> 2023-07-28 22:04:16
>> j*ava.util.concurrent.TimeoutException: Heartbeat of TaskManager with id
>> container_e19_1690528962823_0382_01_05 timed out.*
>> at org.apache.flink.runtime.jobmaster.
>> JobMaster$TaskManagerHeartbeatListener.notifyHeartbeatTimeout(JobMaster
>> .java:1147)
>> at org.apache.flink.runtime.heartbeat.HeartbeatMonitorImpl.run(
>> HeartbeatMonitorImpl.java:109)
>> at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:
>> 511)
>> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
>> at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRunAsync(
>> AkkaRpcActor.java:397)
>> at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(
>> AkkaRpcActor.java:190)
>> at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor
>> .handleRpcMessage(FencedAkkaRpcActor.java:74)
>> at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(
>> AkkaRpcActor.java:152)
>> at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26)
>> at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21)
>> at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123)
>> at akka.japi.pf
>> .UnitCaseStatement.applyOrElse(CaseStatements.scala:21)
>> at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170)
>> at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
>> at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
>> at akka.actor.Actor$class.aroundReceive(Actor.scala:517)
>> at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225)
>> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592)
>> at akka.actor.ActorCell.invoke(ActorCell.scala:561)
>> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258)
>> at akka.dispatch.Mailbox.run(Mailbox.scala:225)
>> at akka.dispatch.Mailbox.exec(Mailbox.scala:235)
>> at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
>> at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool
>> .java:1339)
>> at
>> akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
>> at
>> akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread
>> .java:107)
>>
>> Any suggestions, please share with me.
>>
>> Regards,
>> Nagireddy Y
>>
>

[jira] [Created] (FLINK-33443) Make the test "testWriteComplexType" stable

[Krishna Anandan Ganesan (Jira)]

Krishna Anandan Ganesan created FLINK-33443:
---

 Summary: Make the test "testWriteComplexType" stable
 Key: FLINK-33443
 URL: https://issues.apache.org/jira/browse/FLINK-33443
 Project: Flink
  Issue Type: Bug
  Components: Runtime / State Backends
Reporter: Krishna Anandan Ganesan


We are proposing to make the following test stable:
{code:java}
org.apache.flink.connectors.hive.HiveRunnerITCase.testWriteComplexType{code}



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[jira] [Created] (FLINK-33442) UnsupportedOperationException thrown from RocksDBIncrementalRestoreOperation

[Roman Khachatryan (Jira)]

Roman Khachatryan created FLINK-33442:
-

 Summary: UnsupportedOperationException thrown from 
RocksDBIncrementalRestoreOperation
 Key: FLINK-33442
 URL: https://issues.apache.org/jira/browse/FLINK-33442
 Project: Flink
  Issue Type: Bug
  Components: Runtime / State Backends
Affects Versions: 1.17.1
Reporter: Roman Khachatryan
Assignee: Roman Khachatryan
 Fix For: 1.17.2


When using the new rescaling API, it's possible to get
{code:java}
2023-10-31 18:25:05,179 ERROR 
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder [] - 
Caught unexpected exception.
java.lang.UnsupportedOperationException: null
at java.util.Collections$1.remove(Collections.java:4714) ~[?:?]
at java.util.AbstractCollection.remove(AbstractCollection.java:299) 
~[?:?]
at 
org.apache.flink.runtime.checkpoint.StateObjectCollection.remove(StateObjectCollection.java:105)
 ~[flink-runtime-1.17.1-143.jar:1.17.1-143]
at 
org.apache.flink.contrib.streaming.state.restore.RocksDBIncrementalRestoreOperation.restoreWithRescaling(RocksDBIncrementalRestoreOperation.java:294)
 ~[flink-dist-1.17.1-143.jar:1.17.1-143]
at 
org.apache.flink.contrib.streaming.state.restore.RocksDBIncrementalRestoreOperation.restore(RocksDBIncrementalRestoreOperation.java:167)
 ~[flink-dist-1.17.1-143.jar:1.17.1-143]
at 
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:327)
 ~[flink-dist-1.17.1-143.jar:1.17.1-143]
at 
org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend.createKeyedStateBackend(EmbeddedRocksDBStateBackend.java:512)
 ~[flink-dist-1.17.1-143.jar:1.17.1-143]
at 
org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend.createKeyedStateBackend(EmbeddedRocksDBStateBackend.java:99)
 ~[flink-dist-1.17.1-143.jar:1.17.1-143]
at 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:338)
 ~[flink-dist-1.17.1-143.jar:1.17.1-143]
at 
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:168)
 ~[flink-dist-1.17.1-143.jar:1.17.1-143]
at 
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135)
 ~[flink-dist-1.17.1-143.jar:1.17.1-143]
at 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:355)
 ~[flink-dist-1.17.1-143.jar:1.17.1-143]
at 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:166)
 ~[flink-dist-1.17.1-143.jar:1.17.1-143]
at 
org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:256)
 ~[flink-dist-1.17.1-143.jar:1.17.1-143]
at 
org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.initializeStateAndOpenOperators(RegularOperatorChain.java:106)
 ~[flink-dist-1.17.1-143.jar:1.17.1-143]
at 
org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:735)
 ~[flink-dist-1.17.1-143.jar:1.17.1-143]
at 
org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
 ~[flink-dist-1.17.1-143.jar:1.17.1-143]
at 
org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:710)
 ~[flink-dist-1.17.1-143.jar:1.17.1-143]
at 
org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:676)
 ~[flink-dist-1.17.1-143.jar:1.17.1-143]
at 
org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:952)
 [flink-runtime-1.17.1-143.jar:1.17.1-143]
at 
org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:921) 
[flink-runtime-1.17.1-143.jar:1.17.1-143]
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:745) 
[flink-runtime-1.17.1-143.jar:1.17.1-143]
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562) 
[flink-runtime-1.17.1-143.jar:1.17.1-143]
at java.lang.Thread.run(Thread.java:829) [?:?]
2023-10-31 18:25:05,182 WARN  
org.apache.flink.streaming.api.operators.BackendRestorerProcedure [] - 
Exception while restoring keyed state backend for 
KeyedProcessOperator_353a6b34b8b7f1c1d0fb4616d911049c_(1/2) from alternative 
(1/2), will retry while more alternatives are available.
org.apache.flink.runtime.state.BackendBuildingException: Caught unexpected 
exception.
at 
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:407)
 ~[flink-dist-1.17.1-143.jar:1.17.1-143]
at 

[DISCUSS] Connector releases for Flink 1.18

[Danny Cranmer]

Hey all.

Now Flink 1.18 is released we need to do some connector releases for
integration parity. We can use this thread to start the discussions for
each connector release and spawn separate vote threads. Kafka is done [1]
(thanks Gordon) and AWS connectors are in process [2], I appreciate help
with votes on that one.

Opensearch: Flink 1.18 nightly build passing [3]. I volunteer to be release
manager for this one. I will consolidate 1.0.2 [4] and 1.1.0 [5] into a
single release as 1.1.0.
MongoDB: Flink 1.18 nightly build passing [6]. I volunteer to be release
manager for this one. I will work with Jiabao to get FLINK-33257 merged
into 1.1.0 and release that [7].
GCP Pub Sub: Flink 1.18 nightly build passing [8]. I volunteer to be
release manager for this one. Looks like 3.0.2 is ready to go [9], we
should proceed with this.

ElasticSearch: Flink 1.18 nightly build passing [10]. There are a good
stack of changes ready for 3.1.0 [11], suggest we release that.
JDBC: Flink 1.18 nightly build passing [12]. There are a good stack of
changes ready for 3.2.0 [13], suggest we release that.
RabbitMQ: Flink 1.18 nightly build passing [14]. There are no changes ready
for 3.0.2 [15], recommend we do a minimal 3.0.1-1.18.

Pulsar: The nightly CI is failing [16], needs a deeper look
Cassandra: The nightly CI is failing [17], needs a deeper look

Once I have completed Opensearch/MongoDB/GCP I will pick up others, but
hope others can help out.

Thanks,
Danny

[1] https://lists.apache.org/thread/0lvrm9hl3hnn1fpr74k68lsm22my8xh7
[2] https://lists.apache.org/thread/ko6nrtfsykkz9c9k9392jfj4l9f7qg11
[3] https://github.com/apache/flink-connector-opensearch/actions
[4] https://issues.apache.org/jira/projects/FLINK/versions/12353142
[5] https://issues.apache.org/jira/projects/FLINK/versions/12353141
[6] https://github.com/apache/flink-connector-mongodb/actions
[7] https://issues.apache.org/jira/projects/FLINK/versions/12353483
[8] https://github.com/apache/flink-connector-gcp-pubsub/actions
[9] https://issues.apache.org/jira/projects/FLINK/versions/12353144
[10] https://github.com/apache/flink-connector-elasticsearch/actions
[11] https://issues.apache.org/jira/projects/FLINK/versions/12352520
[12] https://github.com/apache/flink-connector-jdbc/actions
[13] https://issues.apache.org/jira/projects/FLINK/versions/12353143
[14] https://github.com/apache/flink-connector-rabbitmq/actions
[15] https://issues.apache.org/jira/projects/FLINK/versions/12353145
[16] https://github.com/apache/flink-connector-pulsar/actions
[17] https://github.com/apache/flink-connector-cassandra/actions

Re: How do I source debug the scala in the flink table planner?

[Sergey Nuyanzin]

Hi David

thanks on working on it

May be I don't fully understand your issue however if  you are using
IntellijIdea and scala plugin (at least in readme it is mentioned that it
is recommended)
then you should be able to download corresponding sources(at least for
table-planner)  set breakpoints both in java and scala and debug...
Or what is the issue with this?

On Thu, Nov 2, 2023 at 6:59 PM David Radley  wrote:

> Hi,
> I am working on issue https://issues.apache.org/jira/browse/FLINK-33365
> which has been marked as critical and a blocker for the next release of the
> jdbc connector. I can recreate an issue locally using code I built from
> source, so I can add in println’s which are coming out – but this is slow
> and tedious.
>
> Ideally I would like to be able to source debug the scala in the flink
> table planner; any advice would be fab?
>
>  Kind regards, David.
>
> Unless otherwise stated above:
>
> IBM United Kingdom Limited
> Registered in England and Wales with number 741598
> Registered office: PO Box 41, North Harbour, Portsmouth, Hants. PO6 3AU
>


-- 
Best regards,
Sergey

[jira] [Created] (FLINK-33441) Implement union restore tests for ExecUnion node

[Bonnie Varghese (Jira)]

Bonnie Varghese created FLINK-33441:
---

 Summary: Implement union restore tests for ExecUnion node
 Key: FLINK-33441
 URL: https://issues.apache.org/jira/browse/FLINK-33441
 Project: Flink
  Issue Type: Sub-task
Reporter: Bonnie Varghese






--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[jira] [Created] (FLINK-33439) Implement type inference for IN function

[Dawid Wysakowicz (Jira)]

Dawid Wysakowicz created FLINK-33439:


 Summary: Implement type inference for IN function
 Key: FLINK-33439
 URL: https://issues.apache.org/jira/browse/FLINK-33439
 Project: Flink
  Issue Type: Sub-task
  Components: Table SQL / Planner
Reporter: Dawid Wysakowicz
Assignee: Dawid Wysakowicz
 Fix For: 1.19.0






--
This message was sent by Atlassian Jira
(v8.20.10#820010)

How do I source debug the scala in the flink table planner?

[David Radley]

Hi,
I am working on issue https://issues.apache.org/jira/browse/FLINK-33365 which 
has been marked as critical and a blocker for the next release of the jdbc 
connector. I can recreate an issue locally using code I built from source, so I 
can add in println’s which are coming out – but this is slow and tedious.

Ideally I would like to be able to source debug the scala in the flink table 
planner; any advice would be fab?

 Kind regards, David.

Unless otherwise stated above:

IBM United Kingdom Limited
Registered in England and Wales with number 741598
Registered office: PO Box 41, North Harbour, Portsmouth, Hants. PO6 3AU

Re: [RESULT][VOTE] Release flink-connector-hbase v3.0.0, release candidate #2

[Ferenc Csaky]

Hi Martijn!

Is this work in progress?

Thanks,
Ferenc




--- Original Message ---
On Tuesday, September 12th, 2023 at 10:47, Martijn Visser 
 wrote:


> 
> 
> I'm happy to announce that we have unanimously approved this release.
> 
> There are 7 approving votes, 3 of which are binding:
> * Ahmed (non-binding)
> * Sergey (non-binding)
> * Samrat (non-binding)
> * Ferenc (non-binding)
> * Danny (binding)
> * Leonard (binding)
> * Dong (binding)
> 
> There are no disapproving votes.
> 
> I'll work on completing the release. Thanks all!
> 
> Best regards,
> 
> Martijn

[jira] [Created] (FLINK-33440) Bump flink version on flink-connectors-hbase

[Ferenc Csaky (Jira)]

Ferenc Csaky created FLINK-33440:


 Summary: Bump flink version on flink-connectors-hbase
 Key: FLINK-33440
 URL: https://issues.apache.org/jira/browse/FLINK-33440
 Project: Flink
  Issue Type: Improvement
  Components: Connectors / HBase
Reporter: Ferenc Csaky


Follow-up the 1.18 release in the connector repo as well.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[jira] [Created] (FLINK-33438) HiveITCase.testHiveDialect and HiveITCase.testReadWriteHive are failing

[Matthias Pohl (Jira)]

Matthias Pohl created FLINK-33438:
-

 Summary: HiveITCase.testHiveDialect and 
HiveITCase.testReadWriteHive are failing
 Key: FLINK-33438
 URL: https://issues.apache.org/jira/browse/FLINK-33438
 Project: Flink
  Issue Type: Sub-task
  Components: Connectors / Hive, Tests
Reporter: Matthias Pohl


https://github.com/XComp/flink/actions/runs/6729006580/job/18289544587#step:15:12706

{code}
Error: 08:09:00 08:09:00.159 [ERROR] 
org.apache.flink.tests.hive.HiveITCase.testHiveDialect  Time elapsed: 43.377 s  
<<< FAILURE!
Nov 02 08:09:00 org.opentest4j.AssertionFailedError: Did not get expected 
results before timeout, actual result: null. ==> expected:  but was: 

Nov 02 08:09:00 at 
org.junit.jupiter.api.AssertionFailureBuilder.build(AssertionFailureBuilder.java:151)
Nov 02 08:09:00 at 
org.junit.jupiter.api.AssertionFailureBuilder.buildAndThrow(AssertionFailureBuilder.java:132)
Nov 02 08:09:00 at 
org.junit.jupiter.api.AssertTrue.failNotTrue(AssertTrue.java:63)
Nov 02 08:09:00 at 
org.junit.jupiter.api.AssertTrue.assertTrue(AssertTrue.java:36)
Nov 02 08:09:00 at 
org.junit.jupiter.api.Assertions.assertTrue(Assertions.java:211)
Nov 02 08:09:00 at 
org.apache.flink.tests.hive.HiveITCase.checkResultFile(HiveITCase.java:204)
Nov 02 08:09:00 at 
org.apache.flink.tests.hive.HiveITCase.runAndCheckSQL(HiveITCase.java:161)
Nov 02 08:09:00 at 
org.apache.flink.tests.hive.HiveITCase.testHiveDialect(HiveITCase.java:131)
Nov 02 08:09:00 at 
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
[...]
Nov 02 08:09:00 
Error: 08:09:00 08:09:00.159 [ERROR] 
org.apache.flink.tests.hive.HiveITCase.testReadWriteHive  Time elapsed: 37.006 
s  <<< FAILURE!
Nov 02 08:09:00 org.opentest4j.AssertionFailedError: Did not get expected 
results before timeout, actual result: null. ==> expected:  but was: 

Nov 02 08:09:00 at 
org.junit.jupiter.api.AssertionFailureBuilder.build(AssertionFailureBuilder.java:151)
Nov 02 08:09:00 at 
org.junit.jupiter.api.AssertionFailureBuilder.buildAndThrow(AssertionFailureBuilder.java:132)
Nov 02 08:09:00 at 
org.junit.jupiter.api.AssertTrue.failNotTrue(AssertTrue.java:63)
Nov 02 08:09:00 at 
org.junit.jupiter.api.AssertTrue.assertTrue(AssertTrue.java:36)
Nov 02 08:09:00 at 
org.junit.jupiter.api.Assertions.assertTrue(Assertions.java:211)
Nov 02 08:09:00 at 
org.apache.flink.tests.hive.HiveITCase.checkResultFile(HiveITCase.java:204)
Nov 02 08:09:00 at 
org.apache.flink.tests.hive.HiveITCase.runAndCheckSQL(HiveITCase.java:161)
Nov 02 08:09:00 at 
org.apache.flink.tests.hive.HiveITCase.testReadWriteHive(HiveITCase.java:121)
Nov 02 08:09:00 at 
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
[...]
{code}



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

Re: [DISCUSS] FLIP-378: Support Avro timestamp with local timezone

[Martijn Visser]

+1

On Thu, Nov 2, 2023 at 12:44 PM Leonard Xu  wrote:
>
>
> > Thanks @Leonard Xu . Two minor versions are 
> > definitely needed for flip the configs.
>
> Sorry, Peter. I thought the next minor versions are 1.19、2.0， but actually it 
> should be 1.19、1.20、2.0 from current community version plan IIUC, so remove 
> the config in 2.0 should be okay if the 1.20 version exists .
>
> Best,
> Leonard
>
>
> >
> > On Mon, Oct 30, 2023 at 8:55 PM Leonard Xu  > > wrote:
> > Thanks @Peter for driving this FLIP
> >
> > +1 from my side, the timestamp semantics mapping looks good to me.
> >
> > >  In the end, the legacy behavior will be dropped in
> > > Flink 2.0
> > > I don’t think we can drop this option which introduced in 1.19 and drop 
> > > in 2.0, the API removal requires at least two minor versions.
> >
> >
> > Best,
> > Leonard
> >
> > > 2023年10月31日 上午11:18，Peter Huang  > > > 写道：
> > >
> > > Hi Devs,
> > >
> > > Currently, Flink Avro Format doesn't support the Avro time (milli/micros)
> > > with local timezone type.
> > > Although the Avro timestamp (millis/micros) type is supported and is 
> > > mapped
> > > to flink timestamp without timezone,
> > > it is not compliant to semantics defined in Consistent timestamp types in
> > > Hadoop SQL engines
> > >  > >  
> > > >
> > > .
> > >
> > > I propose to support Avro timestamps with the compliance to the mapping
> > > semantics [1] by using a configuration flag.
> > > To keep back compatibility, current mapping is kept as default behavior.
> > > Users can explicitly turn on the new mapping
> > > by setting it to false. In the end, the legacy behavior will be dropped in
> > > Flink 2.0
> > >
> > > Looking forward to your feedback.
> > >
> > >
> > > [1]
> > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-378%3A+Support+Avro+timestamp+with+local+timezone
> > >  
> > > 
> > >
> > >
> > > Best Regards
> > >
> > > Peter Huang
> >
>

Re: [DISCUSS] Confluent Avro support without Schema Registry access

[Martijn Visser]

Hi Dale,

> Aren’t we already fairly dependent on the schema remaining consistent, 
> because otherwise we’d need to update the table schema as well?

No, because the schema can be updated with optional fields and
depending on the compatibility mode, Flink will just consume or
produce nulls in that case.

> I’m not sure what you mean here, sorry. Are you thinking about issues if you 
> needed to mix-and-match with both formatters at the same time? (Rather than 
> just using the Avro formatter as I was describing)

Flink doesn't distinguish a table being a source or a sink. If you
change the Avro format to support reading Schema Registry encoded Avro
format, you would also change it when writing it. However, in order to
write the proper Schema Registry Avro format, you need to have the
magic byte included.

I think the entire point of the Schema Registry Avro messages is that
there is a tight coupling towards a Schema Registry service; that's
the point of the format. I think opening up for alternative processing
is opening up a potential Pandora's box of issues that can be derived
from that: (de)serialization errors, issues with schema evolution
checks as a consumer or a producer etc. I don't see much value for the
Flink project to go in that direction, which would be supporting edge
cases anyway.

Best regards,

Martijn

On Wed, Nov 1, 2023 at 10:36 PM Dale Lane  wrote:
>
> Thanks for the pointer to FLINK-33045 - I hadn’t spotted that. That sounds 
> like it’d address one possible issue (where someone using Flink shouldn’t be, 
> or perhaps doesn’t have access/permission to, register new schemas).
>
> I should be clear that I absolutely agree that using a schema registry is 
> optimum. It should be the norm – it should be the default, preferred and 
> recommended option.
>
> However, I think that there may still be times where the schema registry 
> isn’t available.
>
> Maybe you’re using a mirrored copy of the topic on another kafka cluster and 
> don’t have the original Kafka cluster’s schema registry available. Maybe 
> networking restrictions means where you are running Flink doesn’t have 
> connectivity to the schema registry. Maybe the developer using Flink doesn’t 
> have permission for or access to the schema registry. Maybe the schema 
> registry is currently unavailable. Maybe the developer using Flink is 
> developing their Flink job offline, disconnected from the environment where 
> the schema registry is running (ahead of in future deploying their finished 
> Flink job where it will have access to the schema registry).
>
> It is in such circumstances that I think the approach the avro formatter 
> offers is a useful fallback. Through the table schema, it lets you specify 
> the shape of your data, allowing you to process it without requiring an 
> external dependency.
>
> It seems to me that making it impossible to process Confluent Avro-encoded 
> messages without access to an additional external component is too strict a 
> restriction (as much as there are completely valid reasons for it to be a 
> recommendation).
>
> And, with a very small modification to the avro formatter, it’s a restriction 
> we could remove.
>
> Kind regards
>
> Dale
>
>
>
> From: Ryan Skraba 
> Date: Monday, 30 October 2023 at 16:42
> To: dev@flink.apache.org 
> Subject: [EXTERNAL] Re: [DISCUSS] Confluent Avro support without Schema 
> Registry access
> Hello!  I took a look at FLINK-33045, which is somewhat related: In
> that improvement, the author wants to control who registers schemas.
> The Flink job would know the Avro schema to use, and would look up the
> ID to use in framing the Avro binary.  It uses but never changes the
> schema registry.
>
> Here it sounds like you want nearly the same thing with one more step:
> if the Flink job is configured with the schema to use, it could also
> be pre-configured with the ID that the schema registry knows.
> Technically, it could be configured with a *set* of schemas mapped to
> their IDs when the job starts, but I imagine this would be pretty
> clunky.
>
> I'm curious if you can share what customer use cases wouldn't want
> access to the schema registry!  One of the reasons it exists is to
> prevent systems from writing unreadable or corrupted data to a Kafka
> topic (or other messaging system) -- which I think is what Martijn is
> asking about.  It's unlikely to be a performance gain from hiding it.
>
> Thanks for bringing this up for discussion!  Ryan
>
> [FLINK-33045]: https://issues.apache.org/jira/browse/FLINK-33045
> [Single Object Encoding]:
> https://avro.apache.org/docs/1.11.1/specification/_print/#single-object-encoding-specification
>
> On Fri, Oct 27, 2023 at 3:13 PM Dale Lane  wrote:
> >
> > > if you strip the magic byte, and the schema has
> > > evolved when you're consuming it from Flink,
> > > you can end up with deserialization errors given
> > > that a field might have been deleted/added/
> > > changed etc.
> >
> > Aren’t we already fairly 

Re: [DISCUSS] FLIP-376: Add DISTRIBUTED BY clause

[Martijn Visser]

Hi all,

>From a user point of view, I think it makes sense to go for
DISTRIBUTED BY with how Timo explained it. +1 for his proposal

Best regards,

Martijn


On Thu, Nov 2, 2023 at 11:00 AM Timo Walther  wrote:
>
> Hi Jing,
>
> I agree this is confusing. THe Spark API calls it bucketBy in the
> programmatic API. But anyway, we should discuss the SQL semantics here.
> It's like a "WHERE" is called "filter" in the programmatic world. Or a
> "SELECT" is called "projection" in code.
>
> And looking at all the Hive tutorials[1], distributed by should be more
> consistent. By using the "INTO n BUCKETS", we still include the
> bucketing terminology in the syntax for better understanding.
>
> If there are no other objections to this topic, I would still prefer to
> go with DISTRIBUTED BY.
>
> Regards,
> Timo
>
> [1]
> https://www.linkedin.com/pulse/hive-order-sort-distribute-cluster-mohammad-younus-jameel/
>
>
>
> On 01.11.23 11:55, Jing Ge wrote:
> > Hi Timo,
> >
> > Gotcha, let's use passive verbs. I am actually thinking about "BUCKETED BY
> > 6" or "BUCKETED INTO 6".
> >
> > Not really used in SQL, but afaiu Spark uses the concept[1].
> >
> > [1]
> > https://spark.apache.org/docs/3.1.1/api/python/reference/api/pyspark.sql.DataFrameWriter.bucketBy.html
> >
> >
> > Best regards,
> > Jing
> >
> > On Mon, Oct 30, 2023 at 5:25 PM Timo Walther  wrote:
> >
> >> Hi Jing,
> >>
> >>   > Have you considered using BUCKET BY directly?
> >>
> >> Which vendor uses this syntax? Most vendors that I checked call this
> >> concept "distribution".
> >>
> >> In any case, the "BY" is optional, so certain DDL statements would
> >> declare it like "BUCKET INTO 6 BUCKETS"? And following the PARTITIONED,
> >> we should use the passive voice.
> >>
> >>   > Did you mean users can use their own algorithm? How to do it?
> >>
> >> "own algorithm" only refers to deciding between a list of partitioning
> >> strategies (namely hash and range partitioning) if the connector offers
> >> more than one.
> >>
> >> Regards,
> >> Timo
> >>
> >>
> >> On 30.10.23 12:39, Jing Ge wrote:
> >>> Hi Timo,
> >>>
> >>> The FLIP looks great! Thanks for bringing it to our attention! In order
> >> to
> >>> make sure we are on the same page, I would ask some questions:
> >>>
> >>> 1. DISTRIBUTED BY reminds me DISTRIBUTE BY from Hive like Benchao
> >> mentioned
> >>> which is used to distribute rows amond reducers, i.e. focusing on the
> >>> shuffle during the computation. The FLIP is focusing more on storage, if
> >> I
> >>> am not mistaken. Have you considered using BUCKET BY directly?
> >>>
> >>> 2. According to the FLIP: " CREATE TABLE MyTable (uid BIGINT, name
> >> STRING)
> >>> DISTRIBUTED BY HASH(uid) INTO 6 BUCKETS
> >>>
> >>>  - For advanced users, the algorithm can be defined explicitly.
> >>>  - Currently, either HASH() or RANGE().
> >>>
> >>> "
> >>> Did you mean users can use their own algorithm? How to do it?
> >>>
> >>> Best regards,
> >>> Jing
> >>>
> >>> On Mon, Oct 30, 2023 at 11:13 AM Timo Walther 
> >> wrote:
> >>>
>  Let me reply to the feedback from Yunfan:
> 
> > Distribute by in DML is also supported by Hive
> 
>  I see DISTRIBUTED BY and DISTRIBUTE BY as two separate discussions. This
>  discussion is about DDL. For DDL, we have more freedom as every vendor
>  has custom syntax for CREATE TABLE clauses. Furthermore, this is tightly
>  connector to the connector implementation, not the engine. However, for
>  DML we need to watch out for standard compliance and introduce changes
>  with high caution.
> 
>  How a LookupTableSource interprets the DISTRIBUTED BY is
>  connector-dependent in my opinion. In general this FLIP is a sink
>  ability, but we could have a follow FLIP that helps in distributing load
>  of lookup joins.
> 
> > to avoid data skew problem
> 
>  I understand the use case and that it is important to solve it
>  eventually. Maybe a solution might be to introduce helper Polymorphic
>  Table Functions [1] in the future instead of new syntax.
> 
>  [1]
> 
> 
> >> https://www.ifis.uni-luebeck.de/~moeller/Lectures/WS-19-20/NDBDM/12-Literature/Michels-SQL-2016.pdf
> 
> 
>  Let me reply to the feedback from Benchao:
> 
> > Do you think it's useful to add some extensibility for the hash
>  strategy
> 
>  The hash strategy is fully determined by the connector, not the Flink
>  SQL engine. We are not using Flink's hash strategy in any way. If the
>  hash strategy for the regular Flink file system connector should be
>  changed, this should be expressed via config option. Otherwise we should
>  offer a dedicated `hive-filesystem` or `spark-filesystem` connector.
> 
>  Regards,
>  Timo
> 
> 
>  On 30.10.23 10:44, Timo Walther wrote:
> > Hi Jark,
> >
> > my intention was to avoid too complex syntax in the first version. In
> > the 

Apply to become a Flink contributor

[mzzx]

Hi Guys, 
  I want to contribute to Apache Flink. 
  Would you please give me the permission as a contributor? 
  My JIRA ID is dyccode.

[jira] [Created] (FLINK-33437) Flink 1.17 sink commited legacy Committable state, but it was not removed from state backend

[Yuchi Duan (Jira)]

Yuchi Duan created FLINK-33437:
--

 Summary: Flink 1.17 sink commited legacy Committable state, but it 
was not removed from state backend
 Key: FLINK-33437
 URL: https://issues.apache.org/jira/browse/FLINK-33437
 Project: Flink
  Issue Type: Bug
  Components: Connectors / Kafka, Runtime / Checkpointing
Affects Versions: 1.17.1
 Environment: K8s, Flink 1.17.1
Reporter: Yuchi Duan


My Flink job graph: kafka source -> process -> kafka sink.

I used savepoint to upgrade Flink 1.14.5 to 1.17.1, and the program ran 
normally.A month later, I restarted the Flink job using a savepoint, and the 
job was started normally.Unfortunately, the Flink job failed every time when it 
did a checkpoint.For example the following scenario:
 
 # The program uses Kafka sink
 # Suspend flink job with savepoint A, and Flink Version is 1.14.x
 # Recover the job with savepoint A, and Flink Version is 1.17.1
 # Wait for time longer than {{transactional.id.expiration.ms}} + 
{{transaction.remove.expired.transaction.cleanup.interval.ms}}
 # Suspend flink job with savepoint B, and Flink Version is 1.17.1
 # Recover the job with savepoint B, and Flink Version is 1.17.1
 # Trigger checkpoint ,the Flink job will fail with the following error:
{code:java}
java.io.IOException: Could not perform checkpoint 1009710 for operator 
kafka-sink: Committer (2/2)#2.
    at 
org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:1256)
    at 
org.apache.flink.streaming.runtime.io.checkpointing.CheckpointBarrierHandler.notifyCheckpoint(CheckpointBarrierHandler.java:147)
    at 
org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler.triggerCheckpoint(SingleCheckpointBarrierHandler.java:287)
    at 
org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler.access$100(SingleCheckpointBarrierHandler.java:64)
    at 
org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler$ControllerImpl.triggerGlobalCheckpoint(SingleCheckpointBarrierHandler.java:488)
    at 
org.apache.flink.streaming.runtime.io.checkpointing.AbstractAlignedBarrierHandlerState.triggerGlobalCheckpoint(AbstractAlignedBarrierHandlerState.java:74)
    at 
org.apache.flink.streaming.runtime.io.checkpointing.AbstractAlignedBarrierHandlerState.barrierReceived(AbstractAlignedBarrierHandlerState.java:66)
    at 
org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler.lambda$processBarrier$2(SingleCheckpointBarrierHandler.java:234)
    at 
org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler.markCheckpointAlignedAndTransformState(SingleCheckpointBarrierHandler.java:262)
    at 
org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler.processBarrier(SingleCheckpointBarrierHandler.java:231)
    at 
org.apache.flink.streaming.runtime.io.checkpointing.CheckpointedInputGate.handleEvent(CheckpointedInputGate.java:181)
    at 
org.apache.flink.streaming.runtime.io.checkpointing.CheckpointedInputGate.pollNext(CheckpointedInputGate.java:159)
    at 
org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:118)
    at 
org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
    at 
org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:550)
    at 
org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231)
    at 
org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:839)
    at 
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:788)
    at 
org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:952)
    at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:931)
    at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:745)
    at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562)
    at java.lang.Thread.run(Thread.java:750)
Caused by: 
org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException: 
Could not forward element to next operator
    at 
org.apache.flink.streaming.runtime.tasks.ChainingOutput.pushToOperator(ChainingOutput.java:96)
    at 
org.apache.flink.streaming.runtime.tasks.ChainingOutput.collect(ChainingOutput.java:75)
    at 
org.apache.flink.streaming.runtime.tasks.ChainingOutput.collect(ChainingOutput.java:39)
    at 
org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperator.emit(SinkWriterOperator.java:245)
    at 
org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperator.emitCommittables(SinkWriterOperator.java:215)
    at 

Re: [DISCUSS] FLIP-378: Support Avro timestamp with local timezone

[Leonard Xu]

> Thanks @Leonard Xu . Two minor versions are 
> definitely needed for flip the configs.

Sorry, Peter. I thought the next minor versions are 1.19、2.0， but actually it 
should be 1.19、1.20、2.0 from current community version plan IIUC, so remove the 
config in 2.0 should be okay if the 1.20 version exists .

Best,
Leonard


> 
> On Mon, Oct 30, 2023 at 8:55 PM Leonard Xu  > wrote:
> Thanks @Peter for driving this FLIP
> 
> +1 from my side, the timestamp semantics mapping looks good to me.
> 
> >  In the end, the legacy behavior will be dropped in
> > Flink 2.0
> > I don’t think we can drop this option which introduced in 1.19 and drop in 
> > 2.0, the API removal requires at least two minor versions.
> 
> 
> Best,
> Leonard
> 
> > 2023年10月31日 上午11:18，Peter Huang  > > 写道：
> > 
> > Hi Devs,
> > 
> > Currently, Flink Avro Format doesn't support the Avro time (milli/micros)
> > with local timezone type.
> > Although the Avro timestamp (millis/micros) type is supported and is mapped
> > to flink timestamp without timezone,
> > it is not compliant to semantics defined in Consistent timestamp types in
> > Hadoop SQL engines
> >  >  
> > >
> > .
> > 
> > I propose to support Avro timestamps with the compliance to the mapping
> > semantics [1] by using a configuration flag.
> > To keep back compatibility, current mapping is kept as default behavior.
> > Users can explicitly turn on the new mapping
> > by setting it to false. In the end, the legacy behavior will be dropped in
> > Flink 2.0
> > 
> > Looking forward to your feedback.
> > 
> > 
> > [1]
> > https://cwiki.apache.org/confluence/display/FLINK/FLIP-378%3A+Support+Avro+timestamp+with+local+timezone
> >  
> > 
> > 
> > 
> > Best Regards
> > 
> > Peter Huang
> 

[DISCUSS] FLIP-383: Support Job Recovery for Batch Jobs

[Lijie Wang]

Hi devs,

Zhu Zhu and I would like to start a discussion about FLIP-383: Support Job
Recovery for Batch Jobs[1]

Currently, when Flink’s job manager crashes or gets killed, possibly due to
unexpected errors or planned nodes decommission, it will cause the
following two situations:
1. Failed, if the job does not enable HA.
2. Restart, if the job enable HA. If it’s a streaming job, the job will be
resumed from the last successful checkpoint. If it’s a batch job, it has to
run from beginning, all previous progress will be lost.

In view of this, we think the JM crash may cause great regression for batch
jobs, especially long running batch jobs. This FLIP is mainly to solve this
problem so that batch jobs can recover most job progress after JM crashes.
In this FLIP, our goal is to let most finished tasks not need to be re-run.

You can find more details in the FLIP-383[1]. Looking forward to your
feedback.

[1]
https://cwiki.apache.org/confluence/display/FLINK/FLIP-383%3A+Support+Job+Recovery+for+Batch+Jobs

Best,
Lijie

Re: [DISCUSS] FLIP-376: Add DISTRIBUTED BY clause

[Timo Walther]

Hi Jing,

I agree this is confusing. THe Spark API calls it bucketBy in the 
programmatic API. But anyway, we should discuss the SQL semantics here. 
It's like a "WHERE" is called "filter" in the programmatic world. Or a 
"SELECT" is called "projection" in code.


And looking at all the Hive tutorials[1], distributed by should be more 
consistent. By using the "INTO n BUCKETS", we still include the 
bucketing terminology in the syntax for better understanding.


If there are no other objections to this topic, I would still prefer to 
go with DISTRIBUTED BY.


Regards,
Timo

[1] 
https://www.linkedin.com/pulse/hive-order-sort-distribute-cluster-mohammad-younus-jameel/ 




On 01.11.23 11:55, Jing Ge wrote:

Hi Timo,

Gotcha, let's use passive verbs. I am actually thinking about "BUCKETED BY
6" or "BUCKETED INTO 6".

Not really used in SQL, but afaiu Spark uses the concept[1].

[1]
https://spark.apache.org/docs/3.1.1/api/python/reference/api/pyspark.sql.DataFrameWriter.bucketBy.html


Best regards,
Jing

On Mon, Oct 30, 2023 at 5:25 PM Timo Walther  wrote:


Hi Jing,

  > Have you considered using BUCKET BY directly?

Which vendor uses this syntax? Most vendors that I checked call this
concept "distribution".

In any case, the "BY" is optional, so certain DDL statements would
declare it like "BUCKET INTO 6 BUCKETS"? And following the PARTITIONED,
we should use the passive voice.

  > Did you mean users can use their own algorithm? How to do it?

"own algorithm" only refers to deciding between a list of partitioning
strategies (namely hash and range partitioning) if the connector offers
more than one.

Regards,
Timo


On 30.10.23 12:39, Jing Ge wrote:

Hi Timo,

The FLIP looks great! Thanks for bringing it to our attention! In order

to

make sure we are on the same page, I would ask some questions:

1. DISTRIBUTED BY reminds me DISTRIBUTE BY from Hive like Benchao

mentioned

which is used to distribute rows amond reducers, i.e. focusing on the
shuffle during the computation. The FLIP is focusing more on storage, if

I

am not mistaken. Have you considered using BUCKET BY directly?

2. According to the FLIP: " CREATE TABLE MyTable (uid BIGINT, name

STRING)

DISTRIBUTED BY HASH(uid) INTO 6 BUCKETS

 - For advanced users, the algorithm can be defined explicitly.
 - Currently, either HASH() or RANGE().

"
Did you mean users can use their own algorithm? How to do it?

Best regards,
Jing

On Mon, Oct 30, 2023 at 11:13 AM Timo Walther 

wrote:



Let me reply to the feedback from Yunfan:

   > Distribute by in DML is also supported by Hive

I see DISTRIBUTED BY and DISTRIBUTE BY as two separate discussions. This
discussion is about DDL. For DDL, we have more freedom as every vendor
has custom syntax for CREATE TABLE clauses. Furthermore, this is tightly
connector to the connector implementation, not the engine. However, for
DML we need to watch out for standard compliance and introduce changes
with high caution.

How a LookupTableSource interprets the DISTRIBUTED BY is
connector-dependent in my opinion. In general this FLIP is a sink
ability, but we could have a follow FLIP that helps in distributing load
of lookup joins.

   > to avoid data skew problem

I understand the use case and that it is important to solve it
eventually. Maybe a solution might be to introduce helper Polymorphic
Table Functions [1] in the future instead of new syntax.

[1]



https://www.ifis.uni-luebeck.de/~moeller/Lectures/WS-19-20/NDBDM/12-Literature/Michels-SQL-2016.pdf



Let me reply to the feedback from Benchao:

   > Do you think it's useful to add some extensibility for the hash
strategy

The hash strategy is fully determined by the connector, not the Flink
SQL engine. We are not using Flink's hash strategy in any way. If the
hash strategy for the regular Flink file system connector should be
changed, this should be expressed via config option. Otherwise we should
offer a dedicated `hive-filesystem` or `spark-filesystem` connector.

Regards,
Timo


On 30.10.23 10:44, Timo Walther wrote:

Hi Jark,

my intention was to avoid too complex syntax in the first version. In
the past years, we could enable use cases also without this clause, so
we should be careful with overloading it with too functionality in the
first version. We can still iterate on it later, the interfaces are
flexible enough to support more in the future.

I agree that maybe an explicit HASH and RANGE doesn't harm. Also making
the bucket number optional.

I updated the FLIP accordingly. Now the SupportsBucketing interface
declares more methods that help in validation and proving helpful error
messages to users.

Let me know what you think.

Regards,
Timo


On 27.10.23 10:20, Jark Wu wrote:

Hi Timo,

Thanks for starting this discussion. I really like it!
The FLIP is already in good shape, I only have some minor comments.

1. Could we also support HASH and RANGE distribution kind on the DDL
syntax?
I noticed that HASH and UNKNOWN are introduced in the Java 

[jira] [Created] (FLINK-33436) Documentation on the built-in Profiler

[Yu Chen (Jira)]

Yu Chen created FLINK-33436:
---

 Summary: Documentation on the built-in Profiler
 Key: FLINK-33436
 URL: https://issues.apache.org/jira/browse/FLINK-33436
 Project: Flink
  Issue Type: Sub-task
  Components: Documentation
Affects Versions: 1.19.0
Reporter: Yu Chen






--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[jira] [Created] (FLINK-33435) The visualization and download capabilities of profiling history

[Yu Chen (Jira)]

Yu Chen created FLINK-33435:
---

 Summary: The visualization and download capabilities of profiling 
history 
 Key: FLINK-33435
 URL: https://issues.apache.org/jira/browse/FLINK-33435
 Project: Flink
  Issue Type: Sub-task
  Components: Runtime / Web Frontend
Affects Versions: 1.19.0
Reporter: Yu Chen






--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[jira] [Created] (FLINK-33434) Support invoke async-profiler on Taskmanager through REST API

[Yu Chen (Jira)]

Yu Chen created FLINK-33434:
---

 Summary: Support invoke async-profiler on Taskmanager through REST 
API
 Key: FLINK-33434
 URL: https://issues.apache.org/jira/browse/FLINK-33434
 Project: Flink
  Issue Type: Sub-task
  Components: Runtime / REST
Affects Versions: 1.19.0
Reporter: Yu Chen






--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[jira] [Created] (FLINK-33433) Support invoke async-profiler on Jobmanager through REST API

[Yu Chen (Jira)]

Yu Chen created FLINK-33433:
---

 Summary: Support invoke async-profiler on Jobmanager through REST 
API
 Key: FLINK-33433
 URL: https://issues.apache.org/jira/browse/FLINK-33433
 Project: Flink
  Issue Type: Sub-task
  Components: Runtime / REST
Affects Versions: 1.19.0
Reporter: Yu Chen






--
This message was sent by Atlassian Jira
(v8.20.10#820010)

Re: [DISCUSS] FLIP-381: Deprecate configuration getters/setters that return/set complex Java objects

[weijie guo]

Thanks Junrui for driving this proposal!

I believe this is helpful for the new Process Function API. Because we
don't need to move some related class/components from flink-core to a pure
API module (maybe, called flink-core-api) after this. Even though the FLIP
related to new API is in preparation atm, I still want to emphasize our
goal is that user application should no longer depend on these stuff. So
I'm + 1 for this proposal.


Best regards,

Weijie


Zhu Zhu  于2023年11月2日周四 16:00写道：

> Thanks Junrui for creating the FLIP and kicking off this discussion.
>
> The community has been constantly striving to unify and simplify the
> configuration layer of Flink. Some progress has already been made,
> such as FLINK-29379. However, the compatibility of public interfaces
> poses an obstacle to completing the task. The release of Flink 2.0
> presents a great opportunity to accomplish this goal.
>
> +1 for the proposal.
>
> Thanks,
> Zhu
>
> Rui Fan <1996fan...@gmail.com> 于2023年11月2日周四 10:27写道：
>
> > Thanks Junrui for driving this proposal!
> >
> > ConfigOption is easy to use for flink users, easy to manage options
> > for flink platform maintainers, and easy to maintain for flink developers
> > and flink community.
> >
> > So big +1 for this proposal!
> >
> > Best,
> > Rui
> >
> > On Thu, Nov 2, 2023 at 10:10 AM Junrui Lee  wrote:
> >
> > > Hi devs,
> > >
> > > I would like to start a discussion on FLIP-381: Deprecate configuration
> > > getters/setters that return/set complex Java objects[1].
> > >
> > > Currently, the job configuration in FLINK is spread out across
> different
> > > components, which leads to inconsistencies and confusion. To address
> this
> > > issue, it is necessary to migrate non-ConfigOption complex Java objects
> > to
> > > use ConfigOption and adopt a single Configuration object to host all
> the
> > > configuration.
> > > However, there is a significant blocker in implementing this solution.
> > > These complex Java objects in StreamExecutionEnvironment,
> > CheckpointConfig,
> > > and ExecutionConfig have already been exposed through the public API,
> > > making it challenging to modify the existing implementation.
> > >
> > > Therefore, I propose to deprecate these Java objects and their
> > > corresponding getter/setter interfaces, ultimately removing them in
> > > FLINK-2.0.
> > >
> > > Your feedback and thoughts on this proposal are highly appreciated.
> > >
> > > Best regards,
> > > Junrui Lee
> > >
> > > [1]
> > >
> >
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=278464992
> > >
> >
>

Re: [DISCUSS] FLIP-381: Deprecate configuration getters/setters that return/set complex Java objects

[Zhu Zhu]

Thanks Junrui for creating the FLIP and kicking off this discussion.

The community has been constantly striving to unify and simplify the
configuration layer of Flink. Some progress has already been made,
such as FLINK-29379. However, the compatibility of public interfaces
poses an obstacle to completing the task. The release of Flink 2.0
presents a great opportunity to accomplish this goal.

+1 for the proposal.

Thanks,
Zhu

Rui Fan <1996fan...@gmail.com> 于2023年11月2日周四 10:27写道：

> Thanks Junrui for driving this proposal!
>
> ConfigOption is easy to use for flink users, easy to manage options
> for flink platform maintainers, and easy to maintain for flink developers
> and flink community.
>
> So big +1 for this proposal!
>
> Best,
> Rui
>
> On Thu, Nov 2, 2023 at 10:10 AM Junrui Lee  wrote:
>
> > Hi devs,
> >
> > I would like to start a discussion on FLIP-381: Deprecate configuration
> > getters/setters that return/set complex Java objects[1].
> >
> > Currently, the job configuration in FLINK is spread out across different
> > components, which leads to inconsistencies and confusion. To address this
> > issue, it is necessary to migrate non-ConfigOption complex Java objects
> to
> > use ConfigOption and adopt a single Configuration object to host all the
> > configuration.
> > However, there is a significant blocker in implementing this solution.
> > These complex Java objects in StreamExecutionEnvironment,
> CheckpointConfig,
> > and ExecutionConfig have already been exposed through the public API,
> > making it challenging to modify the existing implementation.
> >
> > Therefore, I propose to deprecate these Java objects and their
> > corresponding getter/setter interfaces, ultimately removing them in
> > FLINK-2.0.
> >
> > Your feedback and thoughts on this proposal are highly appreciated.
> >
> > Best regards,
> > Junrui Lee
> >
> > [1]
> >
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=278464992
> >
>

[jira] [Created] (FLINK-33432) Build artifacts cannot be extracted smoothly

[Matthias Pohl (Jira)]

Matthias Pohl created FLINK-33432:
-

 Summary: Build artifacts cannot be extracted smoothly
 Key: FLINK-33432
 URL: https://issues.apache.org/jira/browse/FLINK-33432
 Project: Flink
  Issue Type: Sub-task
  Components: Build System / CI
Reporter: Matthias Pohl


When extracting the build artifacts there is always a question to overwrite 
certain files:
{code}
[...]
  inflating: logs_117/6_ci  Test (module misc).txt  
replace logs_117/runner-diagnostic-logs/Build UnknownBuildNumber-test.zip? 
[y]es, [n]o, [A]ll, [N]one, [r]ename: y
  inflating: logs_117/runner-diagnostic-logs/Build UnknownBuildNumber-test.zip  
[...]
  inflating: logs_117/1_ci  Test (module python).txt  
replace logs_117/runner-diagnostic-logs/Build UnknownBuildNumber-test.zip? 
[y]es, [n]o, [A]ll, [N]one, [r]ename: y
  inflating: logs_117/runner-diagnostic-logs/Build UnknownBuildNumber-test.zip  
[...]
  inflating: logs_117/4_ci  Test (module connect_2).txt  
replace logs_117/runner-diagnostic-logs/Build UnknownBuildNumber-test.zip? 
[y]es, [n]o, [A]ll, [N]one, [r]ename: y
  inflating: logs_117/runner-diagnostic-logs/Build UnknownBuildNumber-test.zip  
[...]
  inflating: logs_117/3_ci  Test (module connect_1).txt  
replace logs_117/runner-diagnostic-logs/Build UnknownBuildNumber-test.zip? 
[y]es, [n]o, [A]ll, [N]one, [r]ename: y
  inflating: logs_117/runner-diagnostic-logs/Build UnknownBuildNumber-test.zip  
[...]  inflating: logs_117/5_ci  Test (module tests).txt  
replace logs_117/runner-diagnostic-logs/Build UnknownBuildNumber-test.zip? 
[y]es, [n]o, [A]ll, [N]one, [r]ename
[...]
{code}



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[jira] [Created] (FLINK-33431) Create restore tests for ExecNodes

[Dawid Wysakowicz (Jira)]

Dawid Wysakowicz created FLINK-33431:


 Summary: Create restore tests for ExecNodes
 Key: FLINK-33431
 URL: https://issues.apache.org/jira/browse/FLINK-33431
 Project: Flink
  Issue Type: Improvement
  Components: Table SQL / Planner
Reporter: Dawid Wysakowicz
Assignee: Bonnie Varghese
 Fix For: 1.19.0


As a follow up to FLINK-25217 we should create tests for restoring all 
{{ExecNodes}}.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[jira] [Created] (FLINK-33430) Fix Test Failures

[Yang Chen (Jira)]

Yang Chen created FLINK-33430:
-

 Summary: Fix Test Failures
 Key: FLINK-33430
 URL: https://issues.apache.org/jira/browse/FLINK-33430
 Project: Flink
  Issue Type: Bug
Reporter: Yang Chen


This issue is to fix test failures caused by flakiness.

Steps to reproduce test failures:
1)Run the following commands

{{}}
{code:java}

{code}
{{mvn -pl flink-tests edu.illinois:nondex-maven-plugin:2.1.1:nondex 
-Dtest=org.apache.flink.test.misc.GenericTypeInfoTest#testSerializerTree}}

 

2) Then we found the following test failures:
{{}}
{code:java}
[ERROR] GenericTypeInfoTest.testSerializerTree:53 Expected: "GenericTypeInfo 
(PojoWithCollectionGeneric)\n pojos:java.util.List\n key:int\n 
sqlDate:java.sql.Date\n bigInt:java.math.BigInteger\n 
bigDecimalKeepItNull:java.math.BigDecimal\n intVal:java.math.BigInteger\n 
scale:int\n scalaBigInt:scala.math.BigInt\n bigInteger:java.math.BigInteger\n 
mixed:java.util.List\n 
makeMeGeneric:org.apache.flink.test.operators.util.CollectionDataSets$PojoWithDateAndEnum\n
 group:java.lang.String\n date:java.util.Date\n 
cat:org.apache.flink.test.operators.util.CollectionDataSets$Category (is 
enum)\n" but: was "GenericTypeInfo (PojoWithCollectionGeneric)\n key:int\n 
bigDecimalKeepItNull:java.math.BigDecimal\n intVal:java.math.BigInteger\n 
scale:int\n bigInt:java.math.BigInteger\n sqlDate:java.sql.Date\n 
pojos:java.util.List\n scalaBigInt:scala.math.BigInt\n 
bigInteger:java.math.BigInteger\n 
makeMeGeneric:org.apache.flink.test.operators.util.CollectionDataSets$PojoWithDateAndEnum\n
 date:java.util.Date\n 
cat:org.apache.flink.test.operators.util.CollectionDataSets$Category (is 
enum)\n group:java.lang.String\n mixed:java.util.List\n"   {code}
The root cause of the test failure is due to assuming a consistent order of 
{{{}serTree{}}}, but the elements of it are actually not consistent, to fix the 
test failure:

_change the assertion to compare two lists, which can ignore the orders of 
elements_



--
This message was sent by Atlassian Jira
(v8.20.10#820010)