Re: [ANNOUNCE] Apache Flink 1.17.2 released

[Leonard Xu]

Thanks Yun for driving the release.  
Thanks a lot to everyone that has contributed with bug fixes and other 
improvements!

Best,
Leonard


> 2023年11月29日 下午1:05，Yun Tang  写道：
> 
> The Apache Flink community is very happy to announce the release of Apache 
> Flink 1.17.2, which is the second bugfix release for the Apache Flink 1.17 
> series.
> 
> Apache Flink® Is a framework and distributed processing engine for stateful 
> computations over unbounded and bounded data streams. Flink has been designed 
> to run in all common cluster environments, perform computations at in-memory 
> speed and at any scale.
> 
> The release is available for download at:
> https://flink.apache.org/downloads.html 
> 
> 
> Please check out the release blog post for an overview of the improvements 
> for this bugfix release: 
> https://flink.apache.org/2023/11/29/apache-flink-1.17.2-release-announcement/ 
> 
> 
> The full release notes are available in Jira:
> https://issues.apache.org/jira/secure/ReleaseNote.jspa?projectId=12315522=12353260
>  
> 
> 
> We would like to thank all contributors of the Apache Flink community who 
> made this release possible!
> 
> 
> Feel free to reach out to the release managers (or respond to this thread) 
> with feedback on the release process. Our goal is to constantly improve the 
> release process. Feedback on what could be improved or things that didn't go 
> so well are appreciated.
> 
> 
> Regards,
> Release Manager

Re: [ANNOUNCE] Apache Flink 1.17.2 released

[Leonard Xu]

Thanks Yun for driving the release.  
Thanks a lot to everyone that has contributed with bug fixes and other 
improvements!

Best,
Leonard


> 2023年11月29日 下午1:05，Yun Tang  写道：
> 
> The Apache Flink community is very happy to announce the release of Apache 
> Flink 1.17.2, which is the second bugfix release for the Apache Flink 1.17 
> series.
> 
> Apache Flink® Is a framework and distributed processing engine for stateful 
> computations over unbounded and bounded data streams. Flink has been designed 
> to run in all common cluster environments, perform computations at in-memory 
> speed and at any scale.
> 
> The release is available for download at:
> https://flink.apache.org/downloads.html 
> 
> 
> Please check out the release blog post for an overview of the improvements 
> for this bugfix release: 
> https://flink.apache.org/2023/11/29/apache-flink-1.17.2-release-announcement/ 
> 
> 
> The full release notes are available in Jira:
> https://issues.apache.org/jira/secure/ReleaseNote.jspa?projectId=12315522=12353260
>  
> 
> 
> We would like to thank all contributors of the Apache Flink community who 
> made this release possible!
> 
> 
> Feel free to reach out to the release managers (or respond to this thread) 
> with feedback on the release process. Our goal is to constantly improve the 
> release process. Feedback on what could be improved or things that didn't go 
> so well are appreciated.
> 
> 
> Regards,
> Release Manager

[ANNOUNCE] Apache Flink 1.17.2 released

[Yun Tang]

The Apache Flink community is very happy to announce the release of Apache 
Flink 1.17.2, which is the second bugfix release for the Apache Flink 1.17 
series.

Apache Flink® Is a framework and distributed processing engine for stateful 
computations over unbounded and bounded data streams. Flink has been designed 
to run in all common cluster environments, perform computations at in-memory 
speed and at any scale.

The release is available for download at:
https://flink.apache.org/downloads.html

Please check out the release blog post for an overview of the improvements for 
this bugfix release:
https://flink.apache.org/2023/11/29/apache-flink-1.17.2-release-announcement/

The full release notes are available in Jira:
https://issues.apache.org/jira/secure/ReleaseNote.jspa?projectId=12315522=12353260

We would like to thank all contributors of the Apache Flink community who made 
this release possible!


Feel free to reach out to the release managers (or respond to this thread) with 
feedback on the release process. Our goal is to constantly improve the release 
process. Feedback on what could be improved or things that didn't go so well 
are appreciated.


Regards,
Release Manager

[ANNOUNCE] Apache Flink 1.17.2 released

[Yun Tang]

The Apache Flink community is very happy to announce the release of Apache 
Flink 1.17.2, which is the second bugfix release for the Apache Flink 1.17 
series.

Apache Flink® Is a framework and distributed processing engine for stateful 
computations over unbounded and bounded data streams. Flink has been designed 
to run in all common cluster environments, perform computations at in-memory 
speed and at any scale.

The release is available for download at:
https://flink.apache.org/downloads.html

Please check out the release blog post for an overview of the improvements for 
this bugfix release:
https://flink.apache.org/2023/11/29/apache-flink-1.17.2-release-announcement/

The full release notes are available in Jira:
https://issues.apache.org/jira/secure/ReleaseNote.jspa?projectId=12315522=12353260

We would like to thank all contributors of the Apache Flink community who made 
this release possible!


Feel free to reach out to the release managers (or respond to this thread) with 
feedback on the release process. Our goal is to constantly improve the release 
process. Feedback on what could be improved or things that didn't go so well 
are appreciated.


Regards,
Release Manager

Re: Re: inputQueueSize metric

[Feng Jin]

Hi Dimitris

Yes, the inputQueueLength metric can only represent the number of buffers,
and cannot obtain the actual size of the messages.

Regarding the inputQueueSize metric, it ignores the LocalChannel.
Therefore, when tasks are connected through the Forward partition strategy,
it will consistently display a value of 0.
You may consider altering the partition strategy to the rebalance strategy
for testing purposes.st it.


Best,
Feng.


On Tue, Nov 28, 2023 at 2:32 AM Dimitris Banelas via user <
user@flink.apache.org> wrote:

> As per the docs, the `inputQueueLength` metric refers to the number of
> queued input buffers, and cannot be used on its own in order to
> determine buffered records.
>
> For instance, if I know that there are 5 queued input buffers, I cannot
> conclude anything regarding buffered records if the size of each queue
> is not known.
>
>
> On 2023/11/27 17:28:08 Feng Jin wrote:
>  > Hi Dimitris
>  >
>  > Maybe you can use the `inputQueueLength` metric.
>  >
>  >
>  > Best,
>  > Feng
>  >
>  > On Tue, Nov 28, 2023 at 12:07 AM Dimitris Mpanelas via user <
>  > user@flink.apache.org> wrote:
>  >
>  > > Hello,
>  > >
>  > > I am trying to determine the buffered records in the input buffers of
> a
>  > > task. I found the inputQueueSize metric. According to the docs it
> is "The
>  > > real size of queued input buffers in bytes". The docs also state
> that "The
>  > > size for local input channels is always 0 since the local channel take
>  > > records directly from the output queue".
>  > >
>  > > What does this metric truly represent? In which cases should it be
> 0 and
>  > > non 0? Is there a way to determine buffered records (or at least
> something
>  > > similar) by using this metric?
>  > >
>  > > Dimitris.
>  > >
>  > >
>  >
>

Flink listener for Task failures and restarts

[Vignesh Ramesh]

Hi Team,

When a flink job fails we use an implementation of joblistener interface to
get an handle of onJobSubmitted(job submitted) and
onJobExecuted(failure,success). But if my flink application has restart
strategy enabled flink automatically restarts specific task(which fail).
How do we get an handle before/after a task is restarted? We need a custom
logic to be run before/after a task is restarted. Is there any way to
achieve it?

Regards.
Vicky

Re: How to call the open method of JdbcSink?

[Feng Jin]

Hi Sai

I think you can directly cast SinkFunction to GenericJdbcSinkFunction.


https://github.com/apache/flink-connector-jdbc/blob/b477d452ba3aac38d53d1f5d4c4820bdad3ad9cd/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/JdbcSink.java#L63C41-L63C41
```
public static  SinkFunction sink(
String sql,
JdbcStatementBuilder statementBuilder,
JdbcExecutionOptions executionOptions,
JdbcConnectionOptions connectionOptions) {
return new GenericJdbcSinkFunction<>(
new JdbcOutputFormat<>(
new SimpleJdbcConnectionProvider(connectionOptions),
executionOptions,
() -> JdbcBatchStatementExecutor.simple(sql,
statementBuilder)));
}
```


Best,
Feng


On Tue, Nov 28, 2023 at 5:49 PM Sai Vishnu 
wrote:

> Hi team,
>
>
> I am using the JdbcSink from flink-connector-jdbc artifact, version
> 3.1.0-1.17. I am trying to write a Sink wrapper that will internally call
> the invoke method and open method of jdbc sink. While implementing, I see
> that JdbcSink.*sink() *returns a SinkFunction which only exposes the
> invoke method and not the open method.
>
>
> Would appreciate any suggestions on how I can implement this. To add to
> the requirement, the use case is to try and enclose the invoke operation in
> a try catch block so that any exception during the db write process can be
> caught and handled properly.
>
>
> Thanks,
>
> Sai Vishnu Soudri
>

Re: Flink SQL and createRemoteEnvironment

[sangram reddy]

Hi,

createRemoteEnvironment(...) methods have some obscure documentation.

createRemoteEnvironment(String

host,int port,String
...
jarFiles)

What will be the port here? and which jatFiles to supply?

API also mentions "The remote environment sends (parts of) the program
to a cluster for execution. "

What exactly does that mean?

Thanks,

Sangram




On Tue, Nov 28, 2023 at 11:12 AM Xuyang  wrote:

> Hi, Dan.
> Can you provide more details?
>
> > I'm seeing unexpected behavior where it appears like the sql is
> executed locally.
> Did you find a minicluster started locally running you program?
>
> > In my case the remote environment is inside AWS and it doesn't appear
> to pick up the region and credentials unless I set the environment
> variables locally
> I think you need make sure your local machin can connect to AWS
> environment firstly.
>
> Overall, I think `*StreamExecutionEnvironment#**createRemoteEnvironment *`
> can meet your requirements.
>
>
> --
> Best！
> Xuyang
>
>
> At 2023-11-28 03:49:44, "Oxlade, Dan"  wrote:
>
> Hi,
>
> If I use *StreamExecutionEnvironment.createRemoteEnvironment* and then
> *var tEnv = StreamTableEnvironment.create(env)* from the resulting remote 
> *StreamExecutionEvironment
> *will any sql executed using *tEnv.executeSql* be executed remotely
> inside the flink cluster?
>
> I'm seeing unexpected behavior where it appears like the sql is executed
> locally. In my case the remote environment is inside AWS and it doesn't
> appear to pick up the region and credentials unless I set the environment
> variables locally - I want the job to run inside the cluster and use the
> remote AWS context, including things like the auth credentials of the AWS
> task.
>
> I feel like I might be fundamentally misunderstanding.
>
> Thanks
> Dan
>
> T. Rowe Price International Ltd (registered number 3957748) is registered
> in England and Wales with its registered office at Warwick Court, 5
> Paternoster Square, London EC4M 7DX. T. Rowe Price International Ltd is
> authorised and regulated by the Financial Conduct Authority. The company
> has a branch in Dubai International Financial Centre (regulated by the DFSA
> as a Representative Office).
>
> T. Rowe Price (including T. Rowe Price International Ltd and its
> affiliates) and its associates do not provide legal or tax advice. Any
> tax-related discussion contained in this e-mail, including any attachments,
> is not intended or written to be used, and cannot be used, for the purpose
> of (i) avoiding any tax penalties or (ii) promoting, marketing, or
> recommending to any other party any transaction or matter addressed herein.
> Please consult your independent legal counsel and/or professional tax
> advisor regarding any legal or tax issues raised in this e-mail.
>
> The contents of this e-mail and any attachments are intended solely for
> the use of the named addressee(s) and may contain confidential and/or
> privileged information. Any unauthorized use, copying, disclosure, or
> distribution of the contents of this e-mail is strictly prohibited by the
> sender and may be unlawful. If you are not the intended recipient, please
> notify the sender immediately and delete this e-mail.
>
>

Re: Getting a list of tasks for a running job

[Tauseef Janvekar]

Hi Yuxin,

Added flink user group

Thanks,
Tauseef

On Tue, 28 Nov 2023 at 11:38, Tauseef Janvekar 
wrote:

> Hi Yuxin,
> We have deployed it on kubernetes using helm chart -
> https://github.com/bitnami/charts/blob/main/bitnami/flink/values.yaml
> We have used ingress and enabled basic authentication -
> https://kubernetes.github.io/ingress-nginx/examples/auth/basic/
>
> No overrides were done on top of basic helm chart installation.
>
>
> The latest logs are below
> 2023-11-28 06:03:03,636 INFO
>  org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumerator
> [] - Assigning splits to readers {0=[[Partition: aiops-2, StartingOffset:
> 3054, StoppingOffset: -9223372036854775808], [Partition: aiops-0,
> StartingOffset: 2944, StoppingOffset: -9223372036854775808], [Partition:
> aiops-3, StartingOffset: 3006, StoppingOffset: -9223372036854775808], [
> Partition: aiops-1, StartingOffset: 2945, StoppingOffset: -
> 9223372036854775808], [Partition: aiops-4, StartingOffset: 3181,
> StoppingOffset: -9223372036854775808], [Partition: aiops-5, StartingOffset:
> 3154, StoppingOffset: -9223372036854775808]]}
> 2023-11-28 06:03:21,526 INFO  org.apache.pekko.remote.transport.
> ProtocolStateActor [] - No response from remote for outbound
> association. Associate timed out after [2 ms].
> 2023-11-28 06:03:21,539 WARN  org.apache.pekko.remote.
> ReliableDeliverySupervisor   [] - Association with remote system
> [pekko.tcp://flink-metrics@flink-flink-server-taskmanager:35177] has
> failed, address is now gated for [50] ms. Reason: [Association failed
> with [pekko.tcp://flink-metrics@flink-flink-server-taskmanager:35177]]
> Caused by: [No response from remote for outbound association. Associate
> timed out after [2 ms].]
> 2023-11-28 06:03:21,559 WARN  org.apache.pekko.remote.transport.netty.
> NettyTransport   [] - Remote connection to [null] failed with
> org.jboss.netty.channel.ConnectTimeoutException: connection timed out:
> flink-flink-server-taskmanager/172.20.204.52:35177
> 2023-11-28 06:03:43,700 INFO  org.apache.pekko.remote.transport.
> ProtocolStateActor [] - No response from remote for outbound
> association. Associate timed out after [2 ms].
> 2023-11-28 06:03:43,701 WARN  org.apache.pekko.remote.
> ReliableDeliverySupervisor   [] - Association with remote system
> [pekko.tcp://flink-metrics@flink-flink-server-taskmanager:35177] has
> failed, address is now gated for [50] ms. Reason: [Association failed
> with [pekko.tcp://flink-metrics@flink-flink-server-taskmanager:35177]]
> Caused by: [No response from remote for outbound association. Associate
> timed out after [2 ms].]
> 2023-11-28 06:03:43,717 WARN  org.apache.pekko.remote.transport.netty.
> NettyTransport   [] - Remote connection to [null] failed with
> org.jboss.netty.channel.ConnectTimeoutException: connection timed out:
> flink-flink-server-taskmanager/172.20.204.52:35177
> Thanks,
> Tauseef
>
> On Tue, 28 Nov 2023 at 08:07, Yuxin Tan  wrote:
>
>> Hi, Tauseef,
>>
>> AFAIK, the most common way to get a list of tasks that a particular
>> job executes is through Flink's Web UI or REST API.
>>
>> Using the Flink Web UI:
>> When you run a Flink cluster, a Web UI is launched by default on port
>> 8081 of the JobManager. By accessing this Web UI through a browser,
>>  you can see a list of jobs, an overview of each job, and detailed
>> information about a specific job, including the tasks it executes.
>> Using the REST API:
>> For example, to get detailed information about a specific job, you
>> can call the following API: http://:8081/jobs/ [1]
>>
>> > facing the issue where job manager is not able to access task
>> manager but my jobs are completing with no issues.
>> This situation indeed seems peculiar. The limited information
>> provided makes it challenging to pinpoint the exact cause. I would
>> recommend examining the network state, reviewing the configurations,
>> and checking both the JobManager and TaskManager
>> logs for any anomalies or error messages
>>
>> [1]
>> https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/ops/rest_api/
>>
>> Best,
>> Yuxin
>>
>>
>> Tauseef Janvekar  于2023年11月27日周一 19:32写道：
>>
>>> Dear Team,
>>>
>>> How do we get list of tasks that a particular job executes.
>>> If I go toTask Manager then I do not see any tasks. I am also facing the
>>> issue where job manager is not able to access task manager but my jobs are
>>> completing with no issues.
>>>
>>> Any help is appreciated.
>>>
>>> Thanks,
>>> Tauseef
>>>
>>

Re: Job Manager and Task Manager unable to communicate

[Tauseef Janvekar]

Adding flink user group

On Tue, 28 Nov 2023 at 13:39, Tauseef Janvekar 
wrote:

> Did you set some specific job manager or task manager deployment
> parameters ? - No
>
> Did you test without the basic ingress auth ? to be sure this is not
> related to that. - yes we did. And the problem persists.
>
> Please let me know if I can share anything else that might be useful.
>
>
> On Tue, 28 Nov 2023 at 12:52, Benoit Tailhades 
> wrote:
>
>> Did you set some specific job manager or task manager deployment
>> parameters ?
>>
>> Did you test without the basic ingress auth ? to be sure this is not
>> related to that.
>>
>> Le mar. 28 nov. 2023 à 06:58, Tauseef Janvekar 
>> a écrit :
>>
>>> Hi Benoit,
>>>
>>> Are your task manager and job manager on the same vm ?
>>>
>>> We have deployed it on kubernetes cluster with helm chart -
>>> https://github.com/bitnami/charts/blob/main/bitnami/flink/values.yaml
>>> So we cannot confirm if it is on the same vm/node.
>>> One more thing is that we have enabled authentication using basic
>>> ingress auth. -
>>> https://kubernetes.github.io/ingress-nginx/examples/auth/basic/
>>>
>>> How did you configure the Job manager address in the task manager conf
>>> file ?
>>> Did you modify the binding in configuration files ?
>>> It got auto configured using helm chart. We did not modify anything on
>>> top of basic helm chart installation.
>>>
>>> Thanks,
>>> Tauseef
>>>
>>> On Mon, 27 Nov 2023 at 19:29, Benoit Tailhades <
>>> benoit.tailha...@gmail.com> wrote:
>>>
 Hello, Tauseef,

 Can you give more details ? Are your task manager and job manager on
 the same vm ?

 How did you configure the Job manager address in the task manager conf
 file ?
 Did you modify the binding in configuration files ?

 Benoit

 Le lun. 27 nov. 2023 à 14:29, Tauseef Janvekar <
 tauseefjanve...@gmail.com> a écrit :

> Dear Team,
>
> We are getting below error messages in our logs.
> Any help on how to resolve would be greatly appreciated.
>
> 2023-11-27 08:14:29,712 INFO  org.apache.pekko.remote.transport.
> ProtocolStateActor [] - No response from remote for outbound
> association. Associate timed out after [2 ms].
> 2023-11-27 08:14:29,713 WARN  org.apache.pekko.remote.
> ReliableDeliverySupervisor   [] - Association with remote
> system [pekko.tcp://flink-metrics@flink-taskmanager:34309] has
> failed, address is now gated for [50] ms. Reason: [Association failed
> with [pekko.tcp://flink-metrics@flink-taskmanager:34309]] Caused by: [
> No response from remote for outbound association. Associate timed out
> after [2 ms].]
> 2023-11-27 08:14:29,730 WARN  org.apache.pekko.remote.transport.netty.
> NettyTransport   [] - Remote connection to [null] failed with
> org.jboss.netty.channel.ConnectTimeoutException: connection timed
> out: flink-taskmanager/172.20.237.127:34309
> 2023-11-27 08:14:58,401 INFO  org.apache.pekko.remote.transport.
> ProtocolStateActor [] - No response from remote for outbound
> association. Associate timed out after [2 ms].
> 2023-11-27 08:14:58,402 WARN  org.apache.pekko.remote.
> ReliableDeliverySupervisor   [] - Association with remote
> system [pekko.tcp://flink-metrics@flink-taskmanager:34309] has
> failed, address is now gated for [50] ms. Reason: [Association failed
> with [pekko.tcp://flink-metrics@flink-taskmanager:34309]] Caused by: [
> No response from remote for outbound association. Associate timed out
> after [2 ms].]
> 2023-11-27 08:14:58,426 WARN  org.apache.pekko.remote.transport.netty.
> NettyTransport   [] - Remote connection to [null] failed with
> org.jboss.netty.channel.ConnectTimeoutException: connection timed
> out: flink-taskmanager/172.20.237.127:34309
> 2023-11-27 08:15:22,402 INFO  org.apache.pekko.remote.transport.
> ProtocolStateActor [] - No response from remote for outbound
> association. Associate timed out after [2 ms].
> 2023-11-27 08:15:22,403 WARN  org.apache.pekko.remote.
> ReliableDeliverySupervisor   [] - Association with remote
> system [pekko.tcp://flink-metrics@flink-taskmanager:34309] has
> failed, address is now gated for [50] ms. Reason: [Association failed
> with [pekko.tcp://flink-metrics@flink-taskmanager:34309]] Caused by: [
> No response from remote for outbound association. Associate timed out
> after [2 ms].]
> 2023-11-27 08:15:22,434 WARN  org.apache.pekko.remote.transport.netty.
> NettyTransport   [] - Remote connection to [null] failed with
> org.jboss.netty.channel.ConnectTimeoutException: connection timed
> out: flink-taskmanager/172.20.237.127:34309
> 2023-11-27 08:15:46,411 INFO  org.apache.pekko.remote.transport.
> ProtocolStateActor [] - No response from remote for outbound
> 

FlinkSQL environment values in ddl

[Evgeniy Lyutikov]

We started using several flinksql jobs in kubernetes cluster and would like to 
understand how to safely pass passwords and other sensitive data in the 
description of DDLs of tables.

Is there any way to use pointers to environment variables?



"This message contains confidential information/commercial secret. If you are 
not the intended addressee of this message you may not copy, save, print or 
forward it to any third party and you are kindly requested to destroy this 
message and notify the sender thereof by email.
Данное сообщение содержит конфиденциальную информацию/информацию, являющуюся 
коммерческой тайной. Если Вы не являетесь надлежащим адресатом данного 
сообщения, Вы не вправе копировать, сохранять, печатать или пересылать его 
каким либо иным лицам. Просьба уничтожить данное сообщение и уведомить об этом 
отправителя электронным письмом."