Re: Flink Scala Positions in India or USA !

[Martijn Visser]

Hi,

Please don't use the mailing list for this purpose.

Best regards,

Martijn

On Wed, Feb 21, 2024 at 4:08 PM sri hari kali charan Tummala
 wrote:
>
> Hi Folks,
>
> I am currently seeking full-time positions in Flink Scala in India or the USA 
> (non consulting) , specifically at the Principal or Staff level positions in 
> India or USA.
>
> I require an h1b transfer and assistance with relocation from India , my i40 
> is approved.
>
> Thanks & Regards
> Sri Tummala
>

Re: Flink checkpointing - exactly once guaranteed understanding

[Martijn Visser]

Hi Kartik,

It should be the other way around: the connector should use the proper
Source and Sink interfaces, and therefore get the right guarantees and
integration with mechanisms like checkpoints and savepoints. I would
say there's no other way to achieve your desired result, because of
all the edge cases there exist.

Best regards,

Martijn

On Tue, Feb 13, 2024 at 2:08 AM Kartik Kushwaha
 wrote:
>
> Thank you Martijn, the article you provided had detailed explanation on the 
> exactly once two phase commit.
>
>
> Returning to the best way to handle commits/acknowledgments on sources like 
> JMS Queues or Solace topics to  support guaranteed delivery, when they are 
> not supported out of the box by Flink (especially when there is no concept of 
> offset), will the CheckpointListner API and its Checkpoint Subsuming Contract 
> make it possible to perform the JMS message consumption commits by overriding 
> "notifyCheckpointCommit" and "notifyCheckpointAbort" methods. In other words 
> is there a way to find when a Checkpoint is complete in an Operator and 
> perform specific actions when it is complete? Any articles around these would 
> help.
>
> Regards,
> Kartik
>
>
> On Mon, Feb 12, 2024, 10:24 AM Martijn Visser  
> wrote:
>>
>> Sources don't need to support two phase commits, that's something for
>> sinks. I think the example of exactly-once-processing (although the
>> interfaces have changed since) at
>> https://flink.apache.org/2018/02/28/an-overview-of-end-to-end-exactly-once-processing-in-apache-flink-with-apache-kafka-too/
>> is still a good explainer on this topic.
>>
>> Best regards,
>>
>> Martijn
>>
>> On Mon, Feb 12, 2024 at 4:02 PM Kartik Kushwaha
>>  wrote:
>> >
>> > Let me put the question in other words.
>> >
>> > What happens if a source does not support two phase commit and the Flink 
>> > job has to guarantee exactly once delivery to downstream? Checkpointing as 
>> > I understand, works on interval basis. New events for which the checkpoint 
>> > barrier has not yet reached will get dropped or missed. What would be the 
>> > best way to save the state of these non checkpointed data and recover them 
>> > on task crash or job restarts, taking into account that regular checkpoint 
>> > is also enabled and restart and recovery should not lead to duplicates 
>> > from the user managed state vs the checkpointed state.
>> >
>> >
>> > Regards
>> > Kartik
>> >
>> >
>> > On Mon, Feb 12, 2024, 9:50 AM Martijn Visser  
>> > wrote:
>> >>
>> >> Hi Kartik,
>> >>
>> >> I don't think there's much that the Flink community can do here to
>> >> help you. The Solace source and sink aren't owned by the Flink
>> >> project, and based on the source code they haven't been touched for
>> >> the last 7 years [1] and I'm actually not aware of anyone who uses
>> >> Solace at all.
>> >>
>> >> Best regards,
>> >>
>> >> Martijn
>> >>
>> >> [1] 
>> >> https://github.com/SolaceLabs/solace-integration-guides/tree/master/src/flink-jms-connector
>> >>
>> >> On Mon, Feb 12, 2024 at 2:58 PM Kartik Kushwaha
>> >>  wrote:
>> >> >
>> >> > Any help here please.
>> >> >
>> >> > Regards,
>> >> > Kartik
>> >> >
>> >> >
>> >> > On Fri, Feb 9, 2024, 8:33 AM Kartik Kushwaha 
>> >> >  wrote:
>> >> >>
>> >> >> I am using flink checkpointing to restore states of my job. I am using 
>> >> >> unaligned checkpointing with 100 ms as the checkpointing interval. I 
>> >> >> see few events getting dropped that were sucessfully processed by the 
>> >> >> operators or were in-flight that were yet to be captured by 
>> >> >> checkpoint. That is these were new events which came into the pipeline 
>> >> >> between the previously captured checkpoint state and the failure.
>> >> >>
>> >> >> My project acknowledges(commits) back to the topic after the event 
>> >> >> read and mongo ingestion. But the pipeline has transformation, 
>> >> >> enrichment and sink operators after that. These missing events were 
>> >> >> read, ack'd back to the topic and transformed successfully before 
>> >> >> failure and were not yet checkpointed (withing the 100 

Re: Flink checkpointing - exactly once guaranteed understanding

[Martijn Visser]

Sources don't need to support two phase commits, that's something for
sinks. I think the example of exactly-once-processing (although the
interfaces have changed since) at
https://flink.apache.org/2018/02/28/an-overview-of-end-to-end-exactly-once-processing-in-apache-flink-with-apache-kafka-too/
is still a good explainer on this topic.

Best regards,

Martijn

On Mon, Feb 12, 2024 at 4:02 PM Kartik Kushwaha
 wrote:
>
> Let me put the question in other words.
>
> What happens if a source does not support two phase commit and the Flink job 
> has to guarantee exactly once delivery to downstream? Checkpointing as I 
> understand, works on interval basis. New events for which the checkpoint 
> barrier has not yet reached will get dropped or missed. What would be the 
> best way to save the state of these non checkpointed data and recover them on 
> task crash or job restarts, taking into account that regular checkpoint is 
> also enabled and restart and recovery should not lead to duplicates from the 
> user managed state vs the checkpointed state.
>
>
> Regards
> Kartik
>
>
> On Mon, Feb 12, 2024, 9:50 AM Martijn Visser  wrote:
>>
>> Hi Kartik,
>>
>> I don't think there's much that the Flink community can do here to
>> help you. The Solace source and sink aren't owned by the Flink
>> project, and based on the source code they haven't been touched for
>> the last 7 years [1] and I'm actually not aware of anyone who uses
>> Solace at all.
>>
>> Best regards,
>>
>> Martijn
>>
>> [1] 
>> https://github.com/SolaceLabs/solace-integration-guides/tree/master/src/flink-jms-connector
>>
>> On Mon, Feb 12, 2024 at 2:58 PM Kartik Kushwaha
>>  wrote:
>> >
>> > Any help here please.
>> >
>> > Regards,
>> > Kartik
>> >
>> >
>> > On Fri, Feb 9, 2024, 8:33 AM Kartik Kushwaha  
>> > wrote:
>> >>
>> >> I am using flink checkpointing to restore states of my job. I am using 
>> >> unaligned checkpointing with 100 ms as the checkpointing interval. I see 
>> >> few events getting dropped that were sucessfully processed by the 
>> >> operators or were in-flight that were yet to be captured by checkpoint. 
>> >> That is these were new events which came into the pipeline between the 
>> >> previously captured checkpoint state and the failure.
>> >>
>> >> My project acknowledges(commits) back to the topic after the event read 
>> >> and mongo ingestion. But the pipeline has transformation, enrichment and 
>> >> sink operators after that. These missing events were read, ack'd back to 
>> >> the topic and transformed successfully before failure and were not yet 
>> >> checkpointed (withing the 100 ms interval between checkpoints) were 
>> >> dropped.
>> >>
>> >> Pipeline: Source (solace topic, queue reader) --> [MongoWrite + 
>> >> sourcecommit] --> transform --> enrich --> sink (solace topic)
>> >>
>> >> Checkpoint: Unaligned, Exactly-once, 100ms interval, 10ms Min pause 
>> >> between checkpoint
>> >>
>> >> I see that whenever runtime exception is thrown it triggers the close 
>> >> method in each of the functions one by one. Do we have to store the 
>> >> states which were not yet captured by the checkpoint before failure? What 
>> >> happens Network failures or task manager crash or any other abrupt 
>> >> failure?
>> >>
>> >> Or do we have to shift the source topic acknowledgment to the last ( but 
>> >> we will have to chain all this operators to run in a single thread and 
>> >> carry the bytearray message object from solace queue to do ack at the 
>> >> end).
>> >>
>> >> Is there anything else Iam missing here?
>> >>
>> >>
>> >> Note: Sources and Sinks are fully Solace based in all the Flink pipelines 
>> >> ( queues and topics)
>> >>
>> >>

Re: Flink checkpointing - exactly once guaranteed understanding

[Martijn Visser]

Hi Kartik,

I don't think there's much that the Flink community can do here to
help you. The Solace source and sink aren't owned by the Flink
project, and based on the source code they haven't been touched for
the last 7 years [1] and I'm actually not aware of anyone who uses
Solace at all.

Best regards,

Martijn

[1] 
https://github.com/SolaceLabs/solace-integration-guides/tree/master/src/flink-jms-connector

On Mon, Feb 12, 2024 at 2:58 PM Kartik Kushwaha
 wrote:
>
> Any help here please.
>
> Regards,
> Kartik
>
>
> On Fri, Feb 9, 2024, 8:33 AM Kartik Kushwaha  
> wrote:
>>
>> I am using flink checkpointing to restore states of my job. I am using 
>> unaligned checkpointing with 100 ms as the checkpointing interval. I see few 
>> events getting dropped that were sucessfully processed by the operators or 
>> were in-flight that were yet to be captured by checkpoint. That is these 
>> were new events which came into the pipeline between the previously captured 
>> checkpoint state and the failure.
>>
>> My project acknowledges(commits) back to the topic after the event read and 
>> mongo ingestion. But the pipeline has transformation, enrichment and sink 
>> operators after that. These missing events were read, ack'd back to the 
>> topic and transformed successfully before failure and were not yet 
>> checkpointed (withing the 100 ms interval between checkpoints) were dropped.
>>
>> Pipeline: Source (solace topic, queue reader) --> [MongoWrite + 
>> sourcecommit] --> transform --> enrich --> sink (solace topic)
>>
>> Checkpoint: Unaligned, Exactly-once, 100ms interval, 10ms Min pause between 
>> checkpoint
>>
>> I see that whenever runtime exception is thrown it triggers the close method 
>> in each of the functions one by one. Do we have to store the states which 
>> were not yet captured by the checkpoint before failure? What happens Network 
>> failures or task manager crash or any other abrupt failure?
>>
>> Or do we have to shift the source topic acknowledgment to the last ( but we 
>> will have to chain all this operators to run in a single thread and carry 
>> the bytearray message object from solace queue to do ack at the end).
>>
>> Is there anything else Iam missing here?
>>
>>
>> Note: Sources and Sinks are fully Solace based in all the Flink pipelines ( 
>> queues and topics)
>>
>>

[ANNOUNCE] Apache flink-connector-kafka v3.1.0 released

[Martijn Visser]

The Apache Flink community is very happy to announce the release of
Apache flink-connector-kafka v3.1.0. This release is compatible with
Apache Flink 1.17 and 1.18.

Apache Flink® is an open-source stream processing framework for
distributed, high-performing, always-available, and accurate data
streaming applications.

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

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

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

Regards,
Release Manager

[ANNOUNCE] Apache flink-connector-kafka v3.1.0 released

[Martijn Visser]

The Apache Flink community is very happy to announce the release of
Apache flink-connector-kafka v3.1.0. This release is compatible with
Apache Flink 1.17 and 1.18.

Apache Flink® is an open-source stream processing framework for
distributed, high-performing, always-available, and accurate data
streaming applications.

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

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

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

Regards,
Release Manager

Re: Re: Case for a Broadcast join + hint in Flink Streaming SQL

[Martijn Visser]

Hi,

I would definitely expect a FLIP on this topic before moving to
implementation.

Best regards,

Martijn

On Fri, Feb 2, 2024 at 12:47 PM Xuyang  wrote:

> Hi, Prabhjot.
>
> IIUC, the main reasons why the community has not previously considered
> supporting join hints only in batch mode are as follows:
> 1. In batch mode, multiple join type algorithms were implemented quite
> early on, and
> 2. Stream processing represents a long-running scenario, and it is quite
> difficult to determine whether a small table will become a large table
> after a long period of operation.
>
> However, as you mentioned, join hints do indeed have their significance in
> streaming. If you want to support the implementation of "join hints +
> broadcast join" in streaming, the changes I can currently think of include:
> 1. At optimizer, changing the exchange on the small table side to
> broadcast instead of hash (InputProperty#BROADCAST).
> 2. Unknown changes required at the table runtime level.
>
> You can also discuss it within the community through JIRA, FLIP, or the
> dev mailing list.
>
>
> --
> Best！
> Xuyang
>
>
> At 2024-02-02 00:46:01, "Prabhjot Bharaj via user" 
> wrote:
>
> Hi Feng,
>
> Thanks for your prompt response.
> If we were to solve this in Flink, my higher level viewpoint is:
>
> 1. First to implement Broadcast join in Flink Streaming SQL, that works
> across Table api (e.g. via a `left.join(right, ,
> join_type="broadcast")
> 2. Then, support a Broadcast hint that would utilize this new join based
> on the hint type
>
> What do you think about this ?
> Would you have some pointers on how/where to start on the first part ?
> (I'm thinking we'd have to extend the Broadcast state pattern for this
> purpose)
>
> Thanks,
> Prabhjot
>
> On Thu, Feb 1, 2024 at 11:40 AM Feng Jin  wrote:
>
>> Hi Prabhjot
>>
>> I think this is a reasonable scenario. If there is a large table and a
>> very small table for regular join, without broadcasting the regular join,
>> it can easily cause data skew.
>> We have also encountered similar problems too. Currently, we can only
>> copy multiple copies of the small table using the union all and append
>> random values to alleviate data skewness.
>>
>>
>> Best,
>> Feng
>>
>> On Fri, Feb 2, 2024 at 12:24 AM Prabhjot Bharaj via user <
>> user@flink.apache.org> wrote:
>>
>>> Hello folks,
>>>
>>>
>>> We have a use case where we have a few stream-stream joins, requiring us
>>> to join a very large table with a much smaller table, essentially enriching
>>> the large table with a permutation on the smaller table (Consider deriving
>>> all orders/sessions for a new location). Given the nature of the dataset,
>>> if we use a typical join that uses Hash distribution to co-locate the
>>> records for each join key, we end up with a very skewed join (a few task
>>> slots getting all of the work, as against a good distribution).
>>>
>>>
>>> We’ve internally implemented a Salting based solution where we salt the
>>> smaller table and join it with the larger table. While this works in the
>>> POC stage, we’d like to leverage flink as much as possible to do such a
>>> join.
>>>
>>>
>>> By the nature of the problem, a broadcast join seems theoretically
>>> helpful. We’ve done an exploration on query hints supported in Flink,
>>> starting with this FLIP
>>> 
>>> and this FLIP
>>> 
>>> .
>>>
>>>
>>> Currently, the Optimizer doesn't consider the Broadcast hint in the
>>> `Exchange` step of the join, when creating the physical plan (Possibly
>>> because the hint would require the stream-stream join to also support
>>> Broadcast join with SQL)
>>>
>>>
>>> Notice that the Query AST (Abstract Syntax Tree) has the broadcast hint
>>> parsed from the query:
>>>
>>>
>>> ```sql
>>>
>>> ...
>>>
>>> ...
>>>
>>> joinType=[inner], joinHints=[[[BROADCAST inheritPath:[0]
>>> options:[gpla)
>>>
>>> ...
>>>
>>> ```
>>>
>>>
>>> However, the Flink optimizer ignores the hint and still represents the
>>> join as a regular `hash` join in the `Exchange` step:
>>>
>>>
>>> ```sql
>>>
>>> ...
>>>
>>> ...
>>>
>>> :- Exchange(distribution=[hash[shop_id, join_key]])
>>>
>>> ...
>>>
>>> ```
>>>
>>>
>>> In Flink `StreamExecExchange`, the translation happens only via the
>>> `HASH` distribution type
>>> .
>>> unlike in the Flink `BatchExecExchange`, the translation can happen via a
>>> multitude of options
>>> 
>>> (`HASH/BROADCAST`).
>>>
>>>
>>>
>>> Quoting this 

[ANNOUNCE] Community over Code EU 2024 Travel Assistance Applications now open!

[Martijn Visser]

Hi everyone,

The Apache Software Foundation is organizing another Community over Code
event, where a wide variety of speakers will be speaking. You can find all
the details at https://eu.communityovercode.org/

Within the ASF, there is a so-called Travel Assistance Committee (TAC).
This committee exists to help those that would like to attend Community
over Code events, but are unable to do so for financial reasons. I'm hoping
that we'll have a wide variety of Flink community members over there!

All the details and more information can be found in the message below.

Best regards,

Martijn Visser

-- Forwarded message -
From: Christofer Dutz 
Date: Sat, Jan 27, 2024 at 5:31 AM
Subject: Community over Code EU 2024 Travel Assistance Applications now
open!


Hi @,

The Travel Assistance Committee (TAC) are pleased to announce that
travel assistance applications for Community over Code EU 2024 are now
open!

We will be supporting Community over Code EU 2024, Bratislava,
Slovakia, June 3th - 5th, 2024.

TAC exists to help those that would like to attend Community over Code
events, but are unable to do so for financial reasons. For more info
on this years applications and qualifying criteria, please visit the
TAC website at < https://tac.apache.org/ >. Applications are already
open on https://tac-apply.apache.org/, so don't delay!

The Apache Travel Assistance Committee will only be accepting
applications from those people that are able to attend the full event.

Important: Applications close on Friday, March 1st, 2024.

Applicants have until the the closing date above to submit their
applications (which should contain as much supporting material as
required to efficiently and accurately process their request), this
will enable TAC to announce successful applications shortly
afterwards.

As usual, TAC expects to deal with a range of applications from a
diverse range of backgrounds; therefore, we encourage (as always)
anyone thinking about sending in an application to do so ASAP.

We look forward to greeting many of you in Bratislava, Slovakia in June,
2024!

Kind Regards,

Chris

(On behalf of the Travel Assistance Committee)

When replying, please reply to travel-assista...@apache.org

[ANNOUNCE] Community over Code EU 2024 Travel Assistance Applications now open!

[Martijn Visser]

Hi everyone,

The Apache Software Foundation is organizing another Community over Code
event, where a wide variety of speakers will be speaking. You can find all
the details at https://eu.communityovercode.org/

Within the ASF, there is a so-called Travel Assistance Committee (TAC).
This committee exists to help those that would like to attend Community
over Code events, but are unable to do so for financial reasons. I'm hoping
that we'll have a wide variety of Flink community members over there!

All the details and more information can be found in the message below.

Best regards,

Martijn Visser

-- Forwarded message -
From: Christofer Dutz 
Date: Sat, Jan 27, 2024 at 5:31 AM
Subject: Community over Code EU 2024 Travel Assistance Applications now
open!


Hi @,

The Travel Assistance Committee (TAC) are pleased to announce that
travel assistance applications for Community over Code EU 2024 are now
open!

We will be supporting Community over Code EU 2024, Bratislava,
Slovakia, June 3th - 5th, 2024.

TAC exists to help those that would like to attend Community over Code
events, but are unable to do so for financial reasons. For more info
on this years applications and qualifying criteria, please visit the
TAC website at < https://tac.apache.org/ >. Applications are already
open on https://tac-apply.apache.org/, so don't delay!

The Apache Travel Assistance Committee will only be accepting
applications from those people that are able to attend the full event.

Important: Applications close on Friday, March 1st, 2024.

Applicants have until the the closing date above to submit their
applications (which should contain as much supporting material as
required to efficiently and accurately process their request), this
will enable TAC to announce successful applications shortly
afterwards.

As usual, TAC expects to deal with a range of applications from a
diverse range of backgrounds; therefore, we encourage (as always)
anyone thinking about sending in an application to do so ASAP.

We look forward to greeting many of you in Bratislava, Slovakia in June,
2024!

Kind Regards,

Chris

(On behalf of the Travel Assistance Committee)

When replying, please reply to travel-assista...@apache.org

Re: Rabbitmq connector for Flink v1.18

[Martijn Visser]

Hi Charlotta,

I've just pushed out a vote for RabbitMQ connector v3.0.2 which
includes support for Flink 1.18. See
https://lists.apache.org/thread/jmpmrnnwv6yw4ol1zjc5t0frz67jpnqr

Best regards,

Martijn

On Tue, Jan 9, 2024 at 1:08 PM Jiabao Sun  wrote:
>
> Hi Charlotta,
>
> The latest news about connector releases is here[1].
> You can subscribe to the mailing list or follow the jira issue to get the 
> latest updates.
>
> Best,
> Jiabao
>
> [1] https://lists.apache.org/thread/r31f988m57rtjy4s75030pzwrlqybpq2
> [2] https://flink.apache.org/what-is-flink/community/
>
>
> On 2024/01/08 08:55:46 Charlotta Westberg via user wrote:
> > Hi,
> >
> > We are using rabbitmq sources and sinks, and wanted to upgrade to flink 
> > 1.18, but noticed the documentation on RabbitMQ Connector mentioned
> >
> > There is no connector (yet) available for Flink version 1.18
> >
> > I tried to find a JIRA issue for the connector to support flink 1.18 but 
> > was unable to. Is there a plan for the rabbitmq connector and flink 1.18?
> >
> > Best regards
> > Charlotta
> >

Re: Flink 1.17 with Scala 2.13 or Scala 3

[Martijn Visser]

Hi Patrick,

Ideally it would be great if you could provide a small reproducer. It
could be that something isn't working properly in hiding the
internally used version of Scala from users, or something else.
Without a reproducer, it's quite hard to debug.

Best regards,

Martijn

On Wed, Jan 10, 2024 at 2:49 PM Eifler, Patrick  wrote:
>
> Hi Martijn,
>
>
>
> Many thanks for your reply. Yes I have seen the examples. I removed the 
> flink-scala dependency and only use the java libs for everything. So there 
> should no flink-scala API references in the stack.
>
>
>
> These are the flink dependencies we are using:
>
>
>
> "org.apache.flink" % "flink-core" % flinkVersion,
> "org.apache.flink" % "flink-streaming-java" % flinkVersion,
> "org.apache.flink" % "flink-table-api-java" % flinkVersion,
> "org.apache.flink" % "flink-table-api-java-bridge" % flinkVersion,
> "org.apache.flink" % "flink-table-runtime" % flinkVersion,
> "org.apache.flink" % "flink-clients" % flinkVersion,
> "org.apache.flink" % "flink-connector-base" % flinkVersion,
> "org.apache.flink" % "flink-table-planner-loader" % flinkVersion,
> "org.apache.flink" % "flink-connector-kafka" % flinkVersion,
> "org.apache.flink" % "flink-statebackend-rocksdb" % flinkVersion % Provided,
> "org.apache.flink" % "flink-avro" % flinkVersion,
> "org.apache.flink" % "flink-avro-confluent-registry" % flinkVersion,
>
>
>
> And the code path where it fails is not directly related to Flink: (It does 
> not fail on the unit tests).
>
>
>
> Map(
>   PsNowTag -> new OutputTag[T](PsNowTag, typeInformation),
>   SCRNTag -> new OutputTag[T](SCRNTag, typeInformation),
>   GMSTag -> new OutputTag[T](GMSTag, typeInformation),
>   MuleTag -> new OutputTag[T](MuleTag, typeInformation),
>   FutureTag -> new OutputTag[T](FutureTag, typeInformation),
> )
>
>
>
> The failure looks like an incompatibility issue Scala runtime. Scala 2.12 vs. 
> Scala 2.13 collections are not compatible:
>
>
>
> java.lang.NoSuchMethodError: 'scala.collection.immutable.ArraySeq 
> scala.runtime.ScalaRunTime$.wrapRefArray(java.lang.Object[])'
>
>
>
> So this issue seems weird and does look like Flink is using the Scala 2.12 
> runtime even if the flink-scala packages are not installed. The question is 
> why?
>
> Then I would wonder if it is possible to completely exclude Scala from the 
> Flink docker image.
>
>
>
> Best,
>
>
>
> Patrick
>
> --
>
> Patrick Eifler
>
> Staff Software Engineer
>
> FTG Data Engineering
>
> Sony Interactive Entertainment
>
> Kemperplatz 1
>
> 10785 Berlin
>
>
>
> From: Martijn Visser 
> Date: Tuesday, 9. January 2024 at 16:41
> To: Eifler, Patrick 
> Cc: user@flink.apache.org 
> Subject: Re: Flink 1.17 with Scala 2.13 or Scala 3
>
> Hi Patrick,
>
> You're on the right track, because you can't use any of the Flink
> Scala APIs in order to use any arbitrary Scala version. Have you seen
> the examples with Scala 3? [2] Do you have an example of your
> code/setup?
>
> Best regards,
>
> Martijn
>
> [1] https://flink.apache.org/2022/02/22/scala-free-in-one-fifteen/
> [2] https://github.com/sjwiesman/flink-scala-3
>
> On Tue, Jan 9, 2024 at 4:16 PM Eifler, Patrick  
> wrote:
> >
> > Hi,
> >
> >
> >
> > The Flink image still has Scala 2.12 dependencies. I tried to run a flink 
> > job written in Scala 2.13 avoiding the usage of all flink-scala api’s but 
> > getting an incompatibility issue (regarding scala.collections) that I would 
> > normally expect in a project were Scala 2.13 and Scala 2.12 are running 
> > alongside.
> >
> > So I wonder what am I missing, as in the docs and jira tickets everyone 
> > says it works now with any Scala Version.
> > Any pointers are appreciated, Thanks.
> >
> >
> >
> > Best,
> >
> > --
> >
> > Patrick Eifler

Re: Flink release plan

[Martijn Visser]

Hi Praveen,

There have been discussions around an LTS version [1] but no consensus
has yet been reached on that topic.

Best regards,

Martijn

[1] https://lists.apache.org/thread/qvw66of180t3425pnqf2mlx042zhlgnn

On Wed, Jan 10, 2024 at 12:08 PM Praveen Chandna via user
 wrote:
>
> Hello
>
>
>
> Once Flink 2.0 will be released in Dec 2024, what would be the release plan 
> for Flink 1.x ?
>
> Those who are using the Flink 1.20 or earlier releases, do they need to 
> migrate to Flink 2.0 for Bug Fixes or Flink 1.x release track will be active 
> for Bug fixes.
>
>
>
> Thanks !!
>
>
>
> // Regards
>
> Praveen Chandna
>
>

Re: Flink 1.17 with Scala 2.13 or Scala 3

[Martijn Visser]

Hi Patrick,

You're on the right track, because you can't use any of the Flink
Scala APIs in order to use any arbitrary Scala version. Have you seen
the examples with Scala 3? [2] Do you have an example of your
code/setup?

Best regards,

Martijn

[1] https://flink.apache.org/2022/02/22/scala-free-in-one-fifteen/
[2] https://github.com/sjwiesman/flink-scala-3

On Tue, Jan 9, 2024 at 4:16 PM Eifler, Patrick  wrote:
>
> Hi,
>
>
>
> The Flink image still has Scala 2.12 dependencies. I tried to run a flink job 
> written in Scala 2.13 avoiding the usage of all flink-scala api’s but getting 
> an incompatibility issue (regarding scala.collections) that I would normally 
> expect in a project were Scala 2.13 and Scala 2.12 are running alongside.
>
> So I wonder what am I missing, as in the docs and jira tickets everyone says 
> it works now with any Scala Version.
> Any pointers are appreciated, Thanks.
>
>
>
> Best,
>
> --
>
> Patrick Eifler

Re: Continuous Reading of File using FileSource does not process the existing files in version 1.17

[Martijn Visser]

Hi Prasanna,

I think this is as expected. There is no support for monitoring
changes to existing files.

Best regards,

Martijn

On Fri, Jan 5, 2024 at 10:22 AM Prasanna kumar
 wrote:
>
> Hi Flink Community,
>
>
> I hope this email finds you well. I am currently in the process of migrating 
> my Flink application from version 1.12.7 to 1.17.2 and have encountered a 
> behavior issue with the FileSource while reading data from an S3 bucket.
>
>  In the previous version (1.12.7), I was utilizing the readFile method with 
> the TextInputFormat to continuously monitor the S3 bucket for any updates or 
> new files added at a specified time interval. The code snippet for this was 
> as follows:
>
>
>
> streamExecutionEnvironment
> .readFile(new TextInputFormat(new Path("s3://my-s3-path")),
>   "s3://my-s3-path",
>   FileProcessingMode.PROCESS_CONTINUOUSLY,
>   1)
> .setParallelism(1);
>
>
>
> Now, after migrating to Flink 1.17.2, I have switched to using the FileSource 
> for continuous monitoring. The code snippet for this is as follows:
>
> FileSource fileSource = FileSource
> .forRecordStreamFormat(new TextLineInputFormat(), new 
> Path("s3://my-s3-path"))
> .monitorContinuously(Duration.ofMillis(1))
> .build();
>
> streamExecutionEnvironment
> .fromSource(fileSource, WatermarkStrategy.noWatermarks(), "filesource")
> .uid("filesource")
> .setParallelism(1);
>
> While this setup successfully detects new files added to the S3 bucket, but 
> it seems to be missing changes made to existing files. I am unsure if this is 
> expected behavior in Flink 1.17.2 or if there is a configuration detail I 
> might be overlooking.
>
>  Any guidance or suggestions on resolving this issue would be greatly 
> appreciated.
>
> Thanks,
> Prasanna

[DISCUSS] Hadoop 2 vs Hadoop 3 usage

[Martijn Visser]

Hi all,

I want to get some insights on how many users are still using Hadoop 2
vs how many users are using Hadoop 3. Flink currently requires a
minimum version of Hadoop 2.10.2 for certain features, but also
extensively uses Hadoop 3 (like for the file system implementations)

Hadoop 2 has a large number of direct and indirect vulnerabilities
[1]. Most of them can only be resolved by dropping support for Hadoop
2 and upgrading to a Hadoop 3 version. This thread is primarily to get
more insights if Hadoop 2 is still commonly used, or if we can
actually discuss dropping support for Hadoop 2 in Flink.

Best regards,

Martijn

[1] https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-common/2.10.2

Re: Re:Re: Re:Re: Event stuck in the Flink operator

[Martijn Visser]

Hi,

If there's nothing that pushes the watermark forward, then the window
won't be able to close. That's a common thing and expected for every
operator that relies on watermarks. You can also decide to configure
an idleness in order to push the watermark forward if needed.

Best regards,

Martijn

On Tue, Dec 19, 2023 at 9:51 AM T, Yadhunath  wrote:
>
> Hi Xuyang,
>
> Thanks for the reply!
> I haven't used a print connector yet.
>
>
> Thanks,
> Yad
>
> 
> From: Xuyang 
> Sent: Monday, December 18, 2023 8:26 AM
> To: T, Yadhunath 
> Cc: user@flink.apache.org 
> Subject: Re:Re: Re:Re: Event stuck in the Flink operator
>
> This Message Is From an Untrusted Sender
> "Caution" You have not previously corresponded with this sender. If you do 
> not recognize this send, verify their identity offline. If the message 
> appears suspicious, please click the "Report Phish" button in your Outlook 
> client.
>
> Hi, Yad.
> These SQLs seem to be fine. Have you tried using the print connector as a 
> sink to test whether there is any problem?
> If everything goes fine with print sink, then the problem occurs on the kafka 
> sink.
>
>
> --
>
> Best！
> Xuyang
>
>
> 在 2023-12-15 18:48:45，"T, Yadhunath"  写道：
>
> Hi Xuyang,
>
> Thanks for the reply!
>
> I don't have any dataset to share with you at this time.
>
> The last block in the Flink pipeline performs 2 functions -  temporal join( 
> 2nd temporal join) and writing data into the sink topic.
>
> This is what Flink SQL code looks like -
>
> SELECT * FROM
> TABLE1  T1
> LEFT JOIN TABLE2 FOR SYSTEM_TIME AS OF T1.timestamp  as T2  -- 
> Temporal join
> ON *JOIN CONDITION*
> LEFT JOIN TABLE3 FOR SYSTEM_TIME AS OF T1.timestamp  AS T3 -- 
> Temporal join
> ON  *JOIN CONDITION*
>
> Thanks,
> Yad
> 
> From: Xuyang 
> Sent: Friday, December 15, 2023 9:33 AM
> To: user@flink.apache.org 
> Subject: Re:Re: Event stuck in the Flink operator
>
> This Message Is From an Untrusted Sender
> "Caution" You have not previously corresponded with this sender. If you do 
> not recognize this send, verify their identity offline. If the message 
> appears suspicious, please click the "Report Phish" button in your Outlook 
> client.
>
> Hi, Yad.
>
> Can you share the smallest set of sql that can reproduce this problem?
>
> BTW, the last flink operator you mean is the sink with kafka connector?
>
>
>
> --
>
> Best！
> Xuyang
>
>
> 在 2023-12-15 04:38:21，"Alex Cruise"  写道：
>
> Can you share your precise join semantics?
>
> I don't know about Flink SQL offhand, but here are a couple ways to do this 
> when you're using the DataStream API:
>
> * use the Session Window join type, which automatically closes a window after 
> a configurable delay since the last record
> * if you're using a ProcessFunction (very low-level), you can set a timer if 
> you need to react to the non-arrival of data. Just make sure you cancel it 
> when data finally does arrive. :)
>
> -0xe1a
>
> On Thu, Dec 14, 2023 at 6:36 AM T, Yadhunath  
> wrote:
>
>
> Hi,
>
> I am using Flink version 1.16 and I have a streaming job that uses PyFlinkSQL 
> API.
> Whenever a new streaming event comes in it is not getting processed in the 
> last Flink operator  ( it performs temporal join along with writing data into 
> Kafka topic) and it will be only pushed to Kafka on the arrival of the next 
> streaming event. It is like the last operator needs an event to process the 
> previous event. Did anyone experience a similar issue?
> Really appreciate if someone could advise a solution for this.
> Please let me know if you require more input.
>
> Thanks,
> Yad

Re: [ANNOUNCE] Apache Flink Kafka Connectors 3.0.2 released

[Martijn Visser]

Hi Gordon,

Thanks for the release! I've pushed one hotfix [1], to make sure that
the Flink documentation shows the correct version number for the Flink
version it's compatible with.

Best regards,

Martijn

[1] 
https://github.com/apache/flink-connector-kafka/commit/6c3d3d06689336f2fd37bfa5a3b17a5377f07887

On Sat, Dec 2, 2023 at 1:57 AM Tzu-Li (Gordon) Tai  wrote:
>
> The Apache Flink community is very happy to announce the release of Apache 
> Flink Kafka Connectors 3.0.2. This release is compatible with the Apache 
> Flink 1.17.x and 1.18.x release series.
>
> Apache Flink® is an open-source stream processing framework for distributed, 
> high-performing, always-available, and accurate data streaming applications.
>
> The release is available for download at:
> https://flink.apache.org/downloads.html
>
> The full release notes are available in Jira:
> https://issues.apache.org/jira/secure/ReleaseNote.jspa?projectId=12315522=12353768
>
> We would like to thank all contributors of the Apache Flink community who 
> made this release possible!
>
> Regards,
> Gordon

Re: dependency error with latest Kafka connector

[Martijn Visser]

Hi Gordon,

I'm wondering if this might be a difference between how Maven and
Gradle build their projects, since you've done your validations with
Maven, but Günter uses Gradle.
In the end, the quickest fix would be to backport FLINK-30400 to the
Flink Kafka 3.0 release branch.

Best regards,

Martijn

On Fri, Nov 24, 2023 at 12:21 AM Tzu-Li (Gordon) Tai
 wrote:
>
> Hi,
>
> Need to fix my previous comment in the last reply - it should be totally fine 
> that the POM files for flink-connector-kafka 3.0.1-1.18 point to an older 
> version.
> For example, in the ongoing flink-connector-opensearch release 1.1.0-1.18, 
> the POM files also still point to Flink 1.17.1 [1].
>
> If the user intends to compile their job for Flink 1.18.0, then they 
> overwrite the versions for flink-streaming-java and flink-clients accordingly 
> in the user POM.
> The *-1.18 prefix on the externalized connector artifacts simply indicate 
> that they are guaranteed to be compilable and compatible with Flink 1.18.x 
> dependencies.
>
> As a sanity check, I’ve also re-done the tests that were validated during the 
> 3.0.1 release process to see why any issues slipped, but it turns out to be 
> working fine:
>
> built a fat uber jar from quickstart with Flink 1.18.0 for 
> flink-streaming-java and flink-clients, and flink-connector-kafka version 
> 3.0.1-1.18
> then submitted to local Flink cluster 1.18.0. Things worked as expected and 
> the job ran fine.
>
> Now, looking at mvn dependency:tree of the uber jar (see [3] below for the 
> output of dependency:tree), the only Flink dependency being pulled into the 
> user uber jar is flink-connector-base:1.17.0.
>
> FLINK-30400 [2], as Danny pointed out, does intend to address this so that 
> flink-connector-base is not being bundled by user uber jars and should be 
> provided by Flink distributions, but AFAIK there were no breaking changes for 
> the APIs used by classes in flink-connector-base in 1.18.0, so things should 
> still remain compatible (as proven by my local testing).
>
> Which leaves me wondering what was the actual issue that @guenterh.lists 
> bumped into in the first place? Am I missing something obvious?
> Would like to clarify this before I kick off a new release.
>
> Thanks,
> Gordon
>
> [1] 
> https://repository.apache.org/content/repositories/orgapacheflink-1666/org/apache/flink[%E2%80%A6].1.0-1.18/flink-connector-opensearch-parent-1.1.0-1.18.pom
> [2] https://issues.apache.org/jira/browse/FLINK-30400
> [3] mvn dependency:tree output for user job jar:
>
> ```
> [INFO] com.tzulitai:testing-kafka:jar:1.0-SNAPSHOT
> [INFO] +- org.apache.flink:flink-streaming-java:jar:1.18.0:provided
> [INFO] | +- org.apache.flink:flink-core:jar:1.18.0:provided
> [INFO] | | +- org.apache.flink:flink-annotations:jar:1.18.0:provided
> [INFO] | | +- org.apache.flink:flink-metrics-core:jar:1.18.0:provided
> [INFO] | | +- org.apache.flink:flink-shaded-asm-9:jar:9.5-17.0:provided
> [INFO] | | +- org.apache.flink:flink-shaded-jackson:jar:2.14.2-17.0:provided
> [INFO] | | +- org.apache.commons:commons-lang3:jar:3.12.0:provided
> [INFO] | | +- org.apache.commons:commons-text:jar:1.10.0:provided
> [INFO] | | +- com.esotericsoftware.kryo:kryo:jar:2.24.0:provided
> [INFO] | | | +- com.esotericsoftware.minlog:minlog:jar:1.2:provided
> [INFO] | | | \- org.objenesis:objenesis:jar:2.1:provided
> [INFO] | | +- commons-collections:commons-collections:jar:3.2.2:provided
> [INFO] | | \- org.apache.commons:commons-compress:jar:1.21:provided
> [INFO] | +- org.apache.flink:flink-file-sink-common:jar:1.18.0:provided
> [INFO] | +- org.apache.flink:flink-runtime:jar:1.18.0:provided
> [INFO] | | +- org.apache.flink:flink-rpc-core:jar:1.18.0:provided
> [INFO] | | +- org.apache.flink:flink-rpc-akka-loader:jar:1.18.0:provided
> [INFO] | | +- 
> org.apache.flink:flink-queryable-state-client-java:jar:1.18.0:provided
> [INFO] | | +- org.apache.flink:flink-hadoop-fs:jar:1.18.0:provided
> [INFO] | | +- commons-io:commons-io:jar:2.11.0:provided
> [INFO] | | +- 
> org.apache.flink:flink-shaded-netty:jar:4.1.91.Final-17.0:provided
> [INFO] | | +- 
> org.apache.flink:flink-shaded-zookeeper-3:jar:3.7.1-17.0:provided
> [INFO] | | +- org.javassist:javassist:jar:3.24.0-GA:provided
> [INFO] | | +- org.xerial.snappy:snappy-java:jar:1.1.10.4:runtime
> [INFO] | | \- org.lz4:lz4-java:jar:1.8.0:runtime
> [INFO] | +- org.apache.flink:flink-java:jar:1.18.0:provided
> [INFO] | | \- com.twitter:chill-java:jar:0.7.6:provided
> [INFO] | +- org.apache.flink:flink-shaded-guava:jar:31.1-jre-17.0:provided
> [INFO] | +- org.apache.commons:commons-math3:jar:3.6.1:provided
> [INFO] | +- org.slf4j:slf4j-api:jar:1.7.36:runtime
> [INFO] | \- com.google.code.findbugs:jsr305:jar:1.3.9:provided
> [INFO] +- org.apache.flink:flink-clients:jar:1.18.0:provided
> [INFO] | +- org.apache.flink:flink-optimizer:jar:1.18.0:provided
> [INFO] | \- commons-cli:commons-cli:jar:1.5.0:provided
> [INFO] +- org.apache.flink:flink-connector-kafka:jar:3.0.1-1.18:compile

Re: Disable flink old checkpoint clean

[Martijn Visser]

Ah, I actually misread checkpoint and savepoints, sorry. The purpose
of a checkpoint in principle is that Flink manages its lifecycle.
Which S3 interface are you using for the checkpoint storage?

On Tue, Nov 7, 2023 at 6:39 PM Martijn Visser  wrote:
>
> Hi Yang,
>
> If you use the NO_CLAIM mode, Flink will not assume ownership of a
> snapshot and leave it up to the user to delete them. See the blog [1]
> for more details.
>
> Best regards,
>
> Martijn
>
> [1] 
> https://flink.apache.org/2022/05/06/improvements-to-flink-operations-snapshots-ownership-and-savepoint-formats/#no_claim-default-mode
>
> On Tue, Nov 7, 2023 at 5:29 PM Junrui Lee  wrote:
> >
> > Hi Yang,
> >
> >
> > You can try configuring 
> > "execution.checkpointing.externalized-checkpoint-retention: 
> > RETAIN_ON_CANCELLATION"[1] and increasing the value of 
> > "state.checkpoints.num-retained"[2] to retain more checkpoints.
> >
> >
> > Here are the official documentation links for more details:
> >
> > [1]https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/deployment/config/#execution-checkpointing-externalized-checkpoint-retention
> >
> > [2]https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/deployment/config/#state-checkpoints-num-retained
> >
> >
> > Best,
> >
> > Junrui
> >
> >
> > Yang LI  于2023年11月7日周二 22:02写道：
> >>
> >> Dear Flink Community,
> >>
> >> In our Flink application, we persist checkpoints to AWS S3. Recently, 
> >> during periods of high job parallelism and traffic, we've experienced 
> >> checkpoint failures. Upon investigating, it appears these may be related 
> >> to S3 delete object requests interrupting checkpoint re-uploads, as 
> >> evidenced by numerous InterruptedExceptions.
> >>
> >> We aim to explore options for disabling the deletion of stale checkpoints. 
> >> Despite consulting the Flink configuration documentation and conducting 
> >> various tests, the appropriate setting to prevent old checkpoint cleanup 
> >> remains elusive.
> >>
> >> Could you advise if there's a method to disable the automatic cleanup of 
> >> old Flink checkpoints?
> >>
> >> Best,
> >> Yang

Re: Disable flink old checkpoint clean

[Martijn Visser]

Hi Yang,

If you use the NO_CLAIM mode, Flink will not assume ownership of a
snapshot and leave it up to the user to delete them. See the blog [1]
for more details.

Best regards,

Martijn

[1] 
https://flink.apache.org/2022/05/06/improvements-to-flink-operations-snapshots-ownership-and-savepoint-formats/#no_claim-default-mode

On Tue, Nov 7, 2023 at 5:29 PM Junrui Lee  wrote:
>
> Hi Yang,
>
>
> You can try configuring 
> "execution.checkpointing.externalized-checkpoint-retention: 
> RETAIN_ON_CANCELLATION"[1] and increasing the value of 
> "state.checkpoints.num-retained"[2] to retain more checkpoints.
>
>
> Here are the official documentation links for more details:
>
> [1]https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/deployment/config/#execution-checkpointing-externalized-checkpoint-retention
>
> [2]https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/deployment/config/#state-checkpoints-num-retained
>
>
> Best,
>
> Junrui
>
>
> Yang LI  于2023年11月7日周二 22:02写道：
>>
>> Dear Flink Community,
>>
>> In our Flink application, we persist checkpoints to AWS S3. Recently, during 
>> periods of high job parallelism and traffic, we've experienced checkpoint 
>> failures. Upon investigating, it appears these may be related to S3 delete 
>> object requests interrupting checkpoint re-uploads, as evidenced by numerous 
>> InterruptedExceptions.
>>
>> We aim to explore options for disabling the deletion of stale checkpoints. 
>> Despite consulting the Flink configuration documentation and conducting 
>> various tests, the appropriate setting to prevent old checkpoint cleanup 
>> remains elusive.
>>
>> Could you advise if there's a method to disable the automatic cleanup of old 
>> Flink checkpoints?
>>
>> Best,
>> Yang

Re: Issues about removed topics with KafkaSource

[Martijn Visser]

Hi,

That's by design: you can't dynamically add and remove topics from an
existing Flink job that is being restarted from a snapshot. The
feature you're looking for is being planned as part of FLIP-246 [1]

Best regards,

Martijn

[1] https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=217389320


On Wed, Nov 1, 2023 at 7:29 AM Emily Li via user  wrote:
>
> Hey
>
> We have a flinkapp which is subscribing to multiple topics, we recently 
> upgraded our application from 1.13 to 1.15, which we started to use 
> KafkaSource instead of FlinkKafkaConsumer (deprecated).
>
> But we noticed some weird issue with KafkaSource after the upgrade, we are 
> setting the topics with the kafkaSource builder like this
>
> ```
>
> KafkaSource
>
>   .builder[CustomEvent]
>
>   .setBootstrapServers(p.bootstrapServers)
>
>   .setGroupId(consumerGroupName)
>
>   .setDeserializer(deserializer)
>
>   .setTopics(topics)
> ```
>
> And we pass in a list of topics to subscribe, but from time to time we will 
> add some new topics or remove some topics (stop consuming them), but we 
> noticed that ever since we upgraded to 1.15, when we remove a topic from the 
> list, it somehow still consuming the topic (committed offset to the already 
> unsubscribed topics, we also have some logs and metrics showing that we are 
> still consuming the already removed topic), and from the 
> aws.kafka.sum_offset_lag metric, we can also see the removed topic having 
> negative lag...
>
>
> And if we delete the topic in kafka, the running flink application will crash 
> and throw an error "
>
> saying the partition cannot be found (because the topic is already deleted 
> from Kafka).
>
>
> We'd like to understand what could have caused this and if this is a bug in 
> KafkaSource?
>
>
> When we were in 1.13, this never occurred, we were able to remove topics 
> without any issues.
>
>
> We also tried to upgrade to flink 1.17, but the same issue occurred.

Re: [ANNOUNCE] Apache Flink 1.18.0 released

[Martijn Visser]

Thank you all who have contributed!

Op do 26 okt 2023 om 18:41 schreef Feng Jin 

> Thanks for the great work! Congratulations
>
>
> Best,
> Feng Jin
>
> On Fri, Oct 27, 2023 at 12:36 AM Leonard Xu  wrote:
>
> > Congratulations, Well done!
> >
> > Best,
> > Leonard
> >
> > On Fri, Oct 27, 2023 at 12:23 AM Lincoln Lee 
> > wrote:
> >
> > > Thanks for the great work! Congrats all!
> > >
> > > Best,
> > > Lincoln Lee
> > >
> > >
> > > Jing Ge  于2023年10月27日周五 00:16写道：
> > >
> > > > The Apache Flink community is very happy to announce the release of
> > > Apache
> > > > Flink 1.18.0, which is the first release for the Apache Flink 1.18
> > > series.
> > > >
> > > > Apache Flink® is an open-source unified stream and batch data
> > processing
> > > > framework for distributed, high-performing, always-available, and
> > > accurate
> > > > data applications.
> > > >
> > > > 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 release:
> > > >
> > > >
> > >
> >
> https://flink.apache.org/2023/10/24/announcing-the-release-of-apache-flink-1.18/
> > > >
> > > > The full release notes are available in Jira:
> > > >
> > > >
> > >
> >
> https://issues.apache.org/jira/secure/ReleaseNote.jspa?projectId=12315522=12352885
> > > >
> > > > We would like to thank all contributors of the Apache Flink community
> > who
> > > > made this release possible!
> > > >
> > > > Best regards,
> > > > Konstantin, Qingsheng, Sergey, and Jing
> > > >
> > >
> >
>

Re: [ANNOUNCE] Apache Flink 1.18.0 released

[Martijn Visser]

Thank you all who have contributed!

Op do 26 okt 2023 om 18:41 schreef Feng Jin 

> Thanks for the great work! Congratulations
>
>
> Best,
> Feng Jin
>
> On Fri, Oct 27, 2023 at 12:36 AM Leonard Xu  wrote:
>
> > Congratulations, Well done!
> >
> > Best,
> > Leonard
> >
> > On Fri, Oct 27, 2023 at 12:23 AM Lincoln Lee 
> > wrote:
> >
> > > Thanks for the great work! Congrats all!
> > >
> > > Best,
> > > Lincoln Lee
> > >
> > >
> > > Jing Ge  于2023年10月27日周五 00:16写道：
> > >
> > > > The Apache Flink community is very happy to announce the release of
> > > Apache
> > > > Flink 1.18.0, which is the first release for the Apache Flink 1.18
> > > series.
> > > >
> > > > Apache Flink® is an open-source unified stream and batch data
> > processing
> > > > framework for distributed, high-performing, always-available, and
> > > accurate
> > > > data applications.
> > > >
> > > > 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 release:
> > > >
> > > >
> > >
> >
> https://flink.apache.org/2023/10/24/announcing-the-release-of-apache-flink-1.18/
> > > >
> > > > The full release notes are available in Jira:
> > > >
> > > >
> > >
> >
> https://issues.apache.org/jira/secure/ReleaseNote.jspa?projectId=12315522=12352885
> > > >
> > > > We would like to thank all contributors of the Apache Flink community
> > who
> > > > made this release possible!
> > > >
> > > > Best regards,
> > > > Konstantin, Qingsheng, Sergey, and Jing
> > > >
> > >
> >
>

Re: Plans for upgrading Curl with latest 8.4.0

[Martijn Visser]

Hi Ankur,

This is used during CI runs, but it's not bundled/distributed with
Flink itself.

Best regards,

Martijn

On Thu, Oct 12, 2023 at 12:09 PM Singhal, Ankur  wrote:
>
> Hi Matijn,
>
> This is just a reference but we are using it at multiple places.
> https://github.com/apache/flink/blob/master/tools/ci/maven-utils.sh#L59
>
> Although the hostname we are referring here is hardcoded so it can be 
> mitigated.
>
> Thanks and Regards,
> Ankur Singhal
>
> -----Original Message-
> From: Martijn Visser 
> Sent: Thursday, October 12, 2023 3:24 PM
> To: Singhal, Ankur 
> Cc: user@flink.apache.org
> Subject: Re: Plans for upgrading Curl with latest 8.4.0
>
> Hi Ankur,
>
> Where do you see Flink using/bundling Curl?
>
> Best regards,
>
> Martijn
>
> On Wed, Oct 11, 2023 at 9:08 AM Singhal, Ankur  wrote:
> >
> > Hi Team,
> >
> >
> >
> > Do we have any plans to update flink to support Curl 8.4.0 with earlier 
> > versions having severe vulnerabilities?
> >
> >
> >
> > Thanks & Regards,
> >
> > Ankur Singhal
> >
> >

Re: File Source Watermark Issue

[Martijn Visser]

Hi Kirti Dhar,

There isn't really enough information to answer it: are you using
Flink in bounded mode, how have you created your job, what is
appearing in the logs etc.

Best regards,

Martijn

On Mon, Oct 16, 2023 at 7:01 AM Kirti Dhar Upadhyay K via user
 wrote:
>
> Hi Community,
>
>
>
> Can someone help me here?
>
>
>
> Regards,
>
> Kirti Dhar
>
>
>
> From: Kirti Dhar Upadhyay K
> Sent: 10 October 2023 15:52
> To: user@flink.apache.org
> Subject: File Source Watermark Issue
>
>
>
> Hi Team,
>
>
>
> I am using Flink File Source with window aggregator as process function, and 
> stuck with a weird issues.
>
> File source doesn’t seem emitting/progressing the watermarks, whereas if I 
> put a delay (say 100ms) while extracting timestamp from event, it is working 
> fine.
>
>
>
> A bit same thing I found in comments here 
> https://stackoverflow.com/questions/68736330/flink-watermark-not-advancing-at-all-stuck-at-9223372036854775808/68743019#68743019
>
>
>
> Can someone help me here?
>
>
>
> Regards,
>
> Kirti Dhar

Re: Flink 1.17.1 with 1.8 projects

[Martijn Visser]

Hi Patricia,

There's no guarantee of compatibility between different Flink minor
versions and it's not supported. If it works, that can be specific to
this use case and could break at any time. It's up to you to determine
if that is sufficient for you or not.

Best regards,

Martijn

On Mon, Oct 16, 2023 at 9:48 AM patricia lee  wrote:
>
> Hi,
>
> Some of my colleagues are using Flink 1.17.1 server but with projects with 
> Flink 1.8 libraries, so far the projects are working fine without issue for a 
> month now.
>
> Will there be any issue that we are not just aware of, if we continue with 
> this kind of set up env? Appreciate any response.
>
>
> Regards,
> Patricia

Re: Plans for upgrading Curl with latest 8.4.0

[Martijn Visser]

Hi Ankur,

Where do you see Flink using/bundling Curl?

Best regards,

Martijn

On Wed, Oct 11, 2023 at 9:08 AM Singhal, Ankur  wrote:
>
> Hi Team,
>
>
>
> Do we have any plans to update flink to support Curl 8.4.0 with earlier 
> versions having severe vulnerabilities?
>
>
>
> Thanks & Regards,
>
> Ankur Singhal
>
>

Re: Incompatible KafkaProducer version

[Martijn Visser]

Hi Krzysztof,

The bundled Flink Kafka connector for 1.17 uses Kafka 3.2.3, see
https://github.com/apache/flink/blob/release-1.17/flink-connectors/flink-connector-kafka/pom.xml#L38
That's also the case for the externalized Flink Kafka connector v3.0,
see https://github.com/apache/flink-connector-kafka/blob/v3.0/pom.xml#L53
The externalized Flink Kafka connector has been upgraded to use 3.4.0,
but that hasn't been released yet. See
https://github.com/apache/flink-connector-kafka/blob/main/pom.xml#L53
for the details.

Best regards,

Martijn

On Tue, Sep 12, 2023 at 7:36 AM Krzysztof Jankiewicz
 wrote:
>
> Hi, Hang.
>
> There have been a lot of changes made to the TransactionsManager in the Kafka 
> client in 2022.
> (e.g. 
> https://github.com/apache/kafka/commit/3ea7b418fb3d7e9fc74c27751c1b02b04877f197).
>
> Version 3.2.3 was the last one when the TransactionsManager class contained 
> attributes (e.g., topicPartitionBookkeeper) referenced by 
> flink-connector-kafka (1.17.1).
>
> Thanks once again.
> Krzysztof
>
> pon., 11 wrz 2023 o 11:24 Hang Ruan  napisał(a):
>>
>> Hi, Krzysztof.
>>
>> I find that this part has been changed in PR[1] when updating the kafka 
>> client version to 3.4.0.
>> This fix is not released yet. Maybe you can package and check it by yourself.
>>
>> Best,
>> Hang
>>
>> [1] https://github.com/apache/flink-connector-kafka/pull/11
>>
>> Krzysztof Jankiewicz  于2023年9月10日周日 21:52写道：
>>>
>>> Hi,
>>>
>>> I am currently working on a simple application that requires exactly-once 
>>> end-to-end guarantee.
>>>
>>> I am reading data from Kafka and writing it back to Kafka.
>>>
>>> When I use `DeliveryGuarantee.AT_LEAST_ONCE` at the Kafka Sink level, 
>>> everything works fine.
>>> Here's the relevant code:
>>>
>>> KafkaSink sink = KafkaSink.builder()
>>> . . .
>>> .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
>>> . . .
>>> .build();
>>>
>>> Unfortunately, when I switch to DeliveryGuarantee.EXACTLY_ONCE, I encounter 
>>> the following error during error handling (High Availability mode in k8s)::
>>>
>>> Caused by: java.lang.RuntimeException: Incompatible KafkaProducer version
>>>   at 
>>> org.apache.flink.connector.kafka.sink.FlinkKafkaInternalProducer.getField(FlinkKafkaInternalProducer.java:266)
>>>  ~[flink-connector-kafka-1.17.1.jar:1.17.1]
>>> . . .
>>> Caused by: java.lang.NoSuchFieldException: topicPartitionBookkeeper
>>>
>>> The code causing this issue is as follows 
>>> (org.apache.flink.connector.kafka.sink.FlinkKafkaInternalProducer):
>>>
>>> Object transactionManager = this.getTransactionManager();
>>> synchronized(transactionManager) {
>>> Object topicPartitionBookkeeper = getField(transactionManager, 
>>> "topicPartitionBookkeeper");
>>> transitionTransactionManagerStateTo(transactionManager, 
>>> "INITIALIZING");
>>> invoke(topicPartitionBookkeeper, "reset");
>>> setField(transactionManager, "producerIdAndEpoch", 
>>> createProducerIdAndEpoch(producerId, epoch));
>>>
>>> I am using Apache Kafka 1.17.1 and Apache Kafka Client 
>>> (org.apache.kafka:kafka-clients) 3.5.1.
>>> I have examined the code of 
>>> org.apache.kafka.clients.producer.internals.TransactionManager, which is 
>>> used by org.apache.kafka.clients.producer.KafkaProducer.
>>> I can see the producerIdAndEpoch field, but there is no 
>>> topicPartitionBookkeeper field.
>>>
>>> Could you please advise which version of KafkaProducer is compatible with 
>>> the flink-connector-kafka? And am I missing something in my configuration?
>>>
>>> Kind regards
>>> Krzysztof

[CVE-2023-41834] Apache Flink Stateful Functions allowed HTTP header injection due to Improper Neutralization of CRLF Sequences

[Martijn Visser]

CVE-2023-41834: Apache Flink Stateful Functions allowed HTTP header
injection due to Improper Neutralization of CRLF Sequences

Severity: moderate

Vendor:
The Apache Software Foundation

Versions Affected:
Stateful Functions 3.1.0 to 3.2.0

Description:
Improper Neutralization of CRLF Sequences in HTTP Headers in Apache
Flink Stateful Functions 3.1.0, 3.1.1 and 3.2.0 allows remote
attackers to inject arbitrary HTTP headers and conduct HTTP response
splitting attacks via crafted HTTP requests. Attackers could
potentially inject malicious content into the HTTP response that is
sent to the user. This could include injecting a fake login form or
other phishing content, or injecting malicious JavaScript code that
can steal user credentials or perform other malicious actions on the
user's behalf.

Mitigation:
Users should upgrade to 3.3.0

Credit:
This issue was discovered by Andrea Cosentino from Apache Software Foundation

References:
https://flink.apache.org/security/

[CVE-2023-41834] Apache Flink Stateful Functions allowed HTTP header injection due to Improper Neutralization of CRLF Sequences

[Martijn Visser]

CVE-2023-41834: Apache Flink Stateful Functions allowed HTTP header
injection due to Improper Neutralization of CRLF Sequences

Severity: moderate

Vendor:
The Apache Software Foundation

Versions Affected:
Stateful Functions 3.1.0 to 3.2.0

Description:
Improper Neutralization of CRLF Sequences in HTTP Headers in Apache
Flink Stateful Functions 3.1.0, 3.1.1 and 3.2.0 allows remote
attackers to inject arbitrary HTTP headers and conduct HTTP response
splitting attacks via crafted HTTP requests. Attackers could
potentially inject malicious content into the HTTP response that is
sent to the user. This could include injecting a fake login form or
other phishing content, or injecting malicious JavaScript code that
can steal user credentials or perform other malicious actions on the
user's behalf.

Mitigation:
Users should upgrade to 3.3.0

Credit:
This issue was discovered by Andrea Cosentino from Apache Software Foundation

References:
https://flink.apache.org/security/

[ANNOUNCE] Apache Flink Stateful Functions Release 3.3.0 released

[Martijn Visser]

The Apache Flink community is excited to announce the release of
Stateful Functions 3.3.0!

Stateful Functions is a cross-platform stack for building Stateful
Serverless applications, making it radically simpler to develop
scalable, consistent, and elastic distributed applications. This new
release upgrades the Flink runtime to 1.16.2.

Release highlight:
- Upgrade underlying Flink dependency to 1.16.2

Release blogpost:
https://flink.apache.org/2023/09/19/stateful-functions-3.3.0-release-announcement/

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

Java SDK can be found at:
https://search.maven.org/artifact/org.apache.flink/statefun-sdk-java/3.3.0/jar

Python SDK can be found at:
https://pypi.org/project/apache-flink-statefun/

GoLang SDK can be found at:
https://github.com/apache/flink-statefun/tree/statefun-sdk-go/v3.3.0

JavaScript SDK can be found at:
https://www.npmjs.com/package/apache-flink-statefun

Official Docker image for Flink Stateful Functions can be
found at: https://hub.docker.com/r/apache/flink-statefun

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

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

Regards,
Martijn Visser

[ANNOUNCE] Apache Flink Stateful Functions Release 3.3.0 released

[Martijn Visser]

The Apache Flink community is excited to announce the release of
Stateful Functions 3.3.0!

Stateful Functions is a cross-platform stack for building Stateful
Serverless applications, making it radically simpler to develop
scalable, consistent, and elastic distributed applications. This new
release upgrades the Flink runtime to 1.16.2.

Release highlight:
- Upgrade underlying Flink dependency to 1.16.2

Release blogpost:
https://flink.apache.org/2023/09/19/stateful-functions-3.3.0-release-announcement/

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

Java SDK can be found at:
https://search.maven.org/artifact/org.apache.flink/statefun-sdk-java/3.3.0/jar

Python SDK can be found at:
https://pypi.org/project/apache-flink-statefun/

GoLang SDK can be found at:
https://github.com/apache/flink-statefun/tree/statefun-sdk-go/v3.3.0

JavaScript SDK can be found at:
https://www.npmjs.com/package/apache-flink-statefun

Official Docker image for Flink Stateful Functions can be
found at: https://hub.docker.com/r/apache/flink-statefun

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

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

Regards,
Martijn Visser

Re: Custom TCP server socket source

[Martijn Visser via user]

Hi Kamal,

The best starting point would be to look at how to write a custom source
connector. Have a look at
https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/sources/
which also includes links to the various classes that you'll need. Please
let us know what else you've tried already and what issues you've
encountered.

Best regards,

Martijn

On Thu, Jul 27, 2023 at 6:54 AM Kamal Mittal via user 
wrote:

> Hello,
>
>
>
> I need to write “Custom server socket source” which accepts client
> connections over a port.
>
>
>
>1. How to scale it across task managers with parallelism <= no. of
>task managers and with same single port
>2. This is needed w.r.t Kubernetes POD deployment model where each POD
>is running a task manager
>
>
>
> Doing it with in same task manager causes port collision which is expected.
>
>
>
> Rgds,
>
> Kamal
>

Re: Unsubscribe

[Martijn Visser]

Hi,

Please send an email to user-unsubscr...@flink.apache.org in order to be
removed from the User mailing list.

Best regards,

Martijn

On Wed, Jul 26, 2023 at 3:44 AM Lu Weizheng 
wrote:

> Unsubscribe
>

Re: Web.submit.enable=false not working

[Martijn Visser]

Hi,

As documented [1] this option "enables uploading and starting jobs through
the Flink UI (true by default). Please note that even when this is
disabled, session clusters still accept jobs through REST requests (HTTP
calls). This flag only guards the feature to upload jobs in the UI."

It won't hide the button in the UI.

Best regards,

Martijn

[1]
https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/config/

On Wed, Jul 19, 2023 at 10:18 AM patricia lee  wrote:

>
> Hi,
>
> I have this requirement to disable the upload jar button in the web ui.
>
> I added the web.submit.enable=false in my flink-conf.yaml but
> port-forwarding still shows the add New button that enables to upload jar
> file.
>
> I confirmed inside the pod that this property is present and set to false.
>
>
> Version: Flink 1.17.0
> Chrome Incognito
>
>
>

Re: PyFlink SQL from Kafka to Iceberg issues

[Martijn Visser]

 image: "minio/minio:latest"
> container_name: "minio"
> ports:
> - "9001:9001"
> - "9000:9000"
> command:
> - "server"
> - "/data"
> - "--console-address"
> - ":9001"
> volumes:
> - "minio:/data"
> environment:
> MINIO_ROOT_USER: "minio"
> MINIO_ROOT_PASSWORD: "minio123"
> networks:
> default:
> aliases:
> - iceberg.minio
>
> mc:
> depends_on:
> - "minio"
> image: "minio/mc"
> container_name: "mc"
> entrypoint: >
> /bin/sh -c "
> until (/usr/bin/mc config host add minio http://minio:9000 minio
> minio123) do echo "...waiting..." && sleep 1; done;
> /usr/bin/mc rm -r --force minio/iceberg;
> /usr/bin/mc mb minio/iceberg;
> /usr/bin/mc policy set public minio/iceberg;
> tail -f /dev/null
> "
>
> broker:
> image: confluentinc/cp-kafka:7.4.0
> hostname: broker
> container_name: broker
> depends_on:
> - controller
> ports:
> - "9092:9092"
> - "9101:9101"
> environment:
> KAFKA_NODE_ID: 1
> KAFKA_LISTENER_SECURITY_PROTOCOL_MAP:
> 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
> KAFKA_ADVERTISED_LISTENERS:
> 'PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092'
> KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
> KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
> KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
> KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
> KAFKA_JMX_PORT: 9101
> KAFKA_JMX_HOSTNAME: localhost
> KAFKA_PROCESS_ROLES: 'broker'
> KAFKA_CONTROLLER_QUORUM_VOTERS: '2@controller:9093'
> KAFKA_LISTENERS: 'PLAINTEXT://broker:29092,PLAINTEXT_HOST://0.0.0.0:9092'
> KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
> KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
> KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'
> # Replace CLUSTER_ID with a unique base64 UUID using "bin/kafka-storage.sh
> random-uuid"
> # See
> https://docs.confluent.io/kafka/operations-tools/kafka-tools.html#kafka-storage-sh
> CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk'
>
> controller:
> image: confluentinc/cp-kafka:7.4.0
> hostname: controller
> container_name: controller
> ports:
> - "9093:9093"
> - "9102:9102"
> environment:
> KAFKA_NODE_ID: 2
> KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
> KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
> KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
> KAFKA_JMX_PORT: 9102
> KAFKA_JMX_HOSTNAME: localhost
> KAFKA_PROCESS_ROLES: 'controller'
> KAFKA_CONTROLLER_QUORUM_VOTERS: '2@controller:9093'
> KAFKA_LISTENERS: 'CONTROLLER://controller:9093'
> KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
> KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
> KAFKA_LOG_DIRS: '/tmp/kraft-controller-logs'
> # Replace CLUSTER_ID with a unique base64 UUID using "bin/kafka-storage.sh
> random-uuid"
> # See
> https://docs.confluent.io/kafka/operations-tools/kafka-tools.html#kafka-storage-sh
> CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk'
>
> # schema-registry:
> # image: confluentinc/cp-schema-registry:7.4.0
> # hostname: schema-registry
> # container_name: schema-registry
> # depends_on:
> # - broker
> # - controller
> # ports:
> # - "8081:8081"
> # environment:
> # SCHEMA_REGISTRY_HOST_NAME: schema-registry
> # SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'broker:29092'
> # SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081
>
> control-center:
> image: confluentinc/cp-enterprise-control-center:7.4.0
> hostname: control-center
> container_name: control-center
> depends_on:
> - broker
> - controller
> # - schema-registry
> ports:
> - "9021:9021"
> environment:
> CONTROL_CENTER_BOOTSTRAP_SERVERS: 'broker:29092'
> # CONTROL_CENTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081;
> CONTROL_CENTER_REPLICATION_FACTOR: 1
> CONTROL_CENTER_INTERNAL_TOPICS_PARTITIONS: 1
> CONTROL_CENTER_MONITORING_INTERCEPTOR_TOPIC_PARTITIONS: 1
> CONFLUENT_METRICS_TOPIC_REPLICATION: 1
> PORT: 9021
>
> volumes:
> minio: null
>
> networks:
> default:
> name: flinkberg
>
> Best,
>
> Dani
>
> On Thu, Jun 29, 2023 at 9:06 AM Martijn Visser 
> wrote:
>
>> Hi Dani,
>>
>> There are two things that I notice:
>>
>> 1. You're mixing different Flink versions (1.16 and 1.17): all Flink
>> artifacts should be from the same Flink version
>> 2. S3 plugins need to be added to the plugins folder of Flink, because
>> they are loaded via the plugin mechanism. See
>> https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/filesystems/plugins/
>>
>> Bes

Re: Hadoop Error on ECS Fargate

[Martijn Visser]

Hi Mengxi Wang,

Which Flink version are you using?

Best regards,

Martijn

On Thu, Jul 13, 2023 at 3:21 PM Wang, Mengxi X via user <
user@flink.apache.org> wrote:

> Hi community,
>
>
>
> We got this kuerberos error with Hadoop as file system on ECS Fargate
> deployment.
>
>
>
> Caused by: org.apache.hadoop.security.KerberosAuthException: failure to
> login: javax.security.auth.login.LoginException:
> java.lang.NullPointerException: invalid null input: name
>
>
>
> Caused by: javax.security.auth.login.LoginException:
> java.lang.NullPointerException: invalid null input: name
>
>
>
> We don’t actually need Kerberos authentication so I’ve added properties to
> disable Hadoop Kerberos authentication to flink-config.yaml and I can see
> from logs they’ve been picked up. But still the errors persist. Can anybody
> help please?
>
>
>
> Best wishes,
>
> Mengxi Wang
>
>
>
> This message is confidential and subject to terms at:
> https://www.jpmorgan.com/emaildisclaimer including on confidential,
> privileged or legal entity information, malicious content and monitoring of
> electronic messages. If you are not the intended recipient, please delete
> this message and notify the sender immediately. Any unauthorized use is
> strictly prohibited.
>

Re: TCP Socket stream scalability

[Martijn Visser]

Hi Kamal,

It would require you to find a way to create a TCP connection on task
managers where you would only read the assigned part of the TCP connection.
Looking at the protocol itself, that most likely would be an issue. A TCP
connection would also be problematic in case of replays and checkpoint
integration, since you can't roll back to the previous messages of the TCP
connection.

Best regards,

Martijn

On Sat, Jul 15, 2023 at 7:20 AM Kamal Mittal via user 
wrote:

> Hello Community,
>
>
>
> Please share views for below mail.
>
>
>
> Rgds,
>
> Kamal
>
>
>
> *From:* Kamal Mittal via user 
> *Sent:* 14 July 2023 12:55 PM
> *To:* user@flink.apache.org
> *Subject:* TCP Socket stream scalability
>
>
>
> Hello,
>
>
>
> TCP Socket stream can be scaled across task managers similarly to file
> enumerator and source reader below?
>
>
>
> Job is submitted with TCP socket source function and a socket will bind on
> a port once and by a task manager. Is it possible to open socket at job
> manager and then scale / divide the work among task managers similarly to
> below?
>
>
>
>
>
> Rgds,
>
> Kamal
>

Re: Getting rid of getCheckpointLock() method

[Martijn Visser]

Hi Sanket,

Yes, that's correct.

Thanks,

Martijn

On Fri, Jul 7, 2023 at 8:00 PM Sanket Agrawal 
wrote:

> Hello Martijn,
>
> Thank you for your reply. Even for the newer versions of Flink it’s
> recommended to use MailboxExecutor in place of StreamTask’s
> getCheckpointLock() method, right?
>
> Thankyou,
> Sanket
>
> On Fri, Jul 7, 2023 at 5:39 AM Martijn Visser 
> wrote:
>
>> Hi Sanket,
>>
>> Have you read the release notes for Flink 1.11 at
>> https://nightlies.apache.org/flink/flink-docs-release-1.11/release-notes/flink-1.11.html#removal-of-deprecated-streamtaskgetcheckpointlock-flink-12484
>> ?
>> Given that Flink 1.11 is a version that's no longer supported in the
>> Flink community (it's 3 years old at this point), I don't think there will
>> be a lot of options here outside those mentioned in the release notes.
>>
>> Best regards,
>>
>> Martijn
>>
>> On Fri, Jul 7, 2023 at 5:35 AM Sanket Agrawal 
>> wrote:
>>
>>> Hello,
>>>
>>> I'm trying to upgrade flink from 1.8 to 1.11 and StreamTask's
>>> getcheckpointlock method is removed and the new recommendation is to use
>>> MailboxExecutor. Currently we're using it like:
>>>
>>> synchronized(operator.getContainingTask.getCheckpointLock()) {
>>> // perform some operation
>>> }
>>>
>>> The purpose of getCheckpointLock in above code is to get the lock so
>>> that the operation can be performed synchronously without interference from
>>> any other checkpoint.
>>>
>>> Since MailboxExecutor is mainly used for asynchronization, I'm finding
>>> it difficult to convert the above piece of code to something that uses
>>> MailboxExecutor.
>>>
>>> I hope my problem makes sense, and I would really appreciate it if
>>> someone could help me to move forward with this problem.
>>>
>>> Thank you
>>>
>>

Re: Using pre-registered schemas with avro-confluent-registry format is not possible

[Martijn Visser]

Hi Jannik,

Can you still share what are the values you're setting for your properties?
>From the top of my head, you need to set:

value.avro-confluent.properties.auto.register.schemas=false
value.avro-confluent.properties.use.latest.version=true

Best regards,

Martijn

On Tue, Jul 11, 2023 at 4:43 AM Schmeier, Jannik 
wrote:

> Do you know how they are overriding the method? Are they building their
> own distribution of Flink with their own implementation of that method? I'd
> like to avoid having to build that myself. I'd be interested in a solution
> in the official release.
>
>
>
> *Von:* Meissner, Dylan 
> *Gesendet:* Freitag, 30. Juni 2023 17:26
> *An:* Martijn Visser ; Schmeier, Jannik
> 
> *Cc:* Schwalbe Matthias ;
> user@flink.apache.org
> *Betreff:* Re: Using pre-registered schemas with avro-confluent-registry
> format is not possible
>
>
>
> This comes up when the Schema Registry has authorization capabilities.
>
>
>
> A producing application may not be authorized to call “register” when
> another principal owns the schema subject, for example in organizations
> with strong schema governance practices. When
> ConfluentSchemaRegistryCoder.writeSchema is invoked, these producers will
> receive HTTP 403 failures.
>
>
>
> In these unfortunate cases, they are overriding writeSchema…
>
>
>
> @Override
>
> public void writeSchema(Schema schema, OutputStream out) throws
> IOException {
>
> int registeredId = schemaRegistryClient.register(subject, schema);
>
> out.write(CONFLUENT_MAGIC_BYTE);
>
> byte[] schemaIdBytes =
> ByteBuffer.allocate(4).putInt(registeredId).array();
>
> out.write(schemaIdBytes);
>
> }
>
>
>
> …by replacing “register()” with the schema lookup call, “getId()”
>
>
>
> @Override
>
> public void writeSchema(Schema schema, OutputStream out) throws
> IOException {
>
> int registeredId = schemaRegistryClient.getId(subject, schema);
>
>     out.write(CONFLUENT_MAGIC_BYTE);
>
> byte[] schemaIdBytes =
> ByteBuffer.allocate(4).putInt(registeredId).array();
>
> out.write(schemaIdBytes);
>
> }
>
>
>
> Dylan
>
>
>
>
>
> *From: *Martijn Visser 
> *Date: *Monday, June 5, 2023 at 7:45 AM
> *To: *Schmeier, Jannik 
> *Cc: *Schwalbe Matthias ,
> user@flink.apache.org 
> *Subject: *Re: Using pre-registered schemas with avro-confluent-registry
> format is not possible
>
> *Caution:* EXTERNAL EMAIL ALERT. This sender is not from your
> organization. If you do not recognize the sender, DO NOT CLICK on links or
> attachments.
>
>
>
> Hi Jannik,
>
>
>
> Can you share how you've set those properties, because I've been able to
> use this without any problems.
>
>
>
> Best regards,
>
>
>
> Martijn
>
>
>
> On Thu, Jun 1, 2023 at 2:43 PM Schmeier, Jannik 
> wrote:
>
> Hello Thias,
>
>
>
> thank you for your answer.
>
>
>
> We've tested registering an existing (byte equal) schema a second time,
> but unfortunately the schema registry does still deny the request.
>
>
>
> Your last suggestion sounds promising, but I think there are some edge
> cases with this approach that will still cause an error. For example when
> writing to a new topic that’s empty, querying this topic before won't
> return any records and therefore the schema would not be put into the
> schemaRegistryClient cache.
>
>
>
> I'm still preferring a flag for the "avro-confluent-registry" format that
> will disable registering schemas and instead the format will just try to
> get the ID for a schema string from the registry. If there is an ID for
> that schema, Flink will use it. If there is none, an exception should be
> thrown.
>
> What do you think of that?
>
>
>
> Best regards,
>
> Jannik
>
>
>
>
>
> *Von:* Schwalbe Matthias 
> *Gesendet:* Mittwoch, 31. Mai 2023 13:33
> *An:* Schmeier, Jannik ; user@flink.apache.org
> *Betreff:* RE: Using pre-registered schemas with avro-confluent-registry
> format is not possible
>
>
>
>
>
> Hello Jannik,
>
>
>
> Some things to consider (I had a similar problem a couple of years before):
>
>- The schemaRegistryClient actually caches schema ids, so it will hit
>the schema registry only once,
>- The schema registered in schema registry needs to be byte-equal,
>otherwise schema registry considers it to be a new schema (version)
>- … to my best knowledge writing an existing schema to the schema
>registry does not fail because it is actually not written
>
>
>- Could be that this is not entirely true as we had to replac

Re: Getting rid of getCheckpointLock() method

[Martijn Visser]

Hi Sanket,

Have you read the release notes for Flink 1.11 at
https://nightlies.apache.org/flink/flink-docs-release-1.11/release-notes/flink-1.11.html#removal-of-deprecated-streamtaskgetcheckpointlock-flink-12484
?
Given that Flink 1.11 is a version that's no longer supported in the Flink
community (it's 3 years old at this point), I don't think there will be a
lot of options here outside those mentioned in the release notes.

Best regards,

Martijn

On Fri, Jul 7, 2023 at 5:35 AM Sanket Agrawal 
wrote:

> Hello,
>
> I'm trying to upgrade flink from 1.8 to 1.11 and StreamTask's
> getcheckpointlock method is removed and the new recommendation is to use
> MailboxExecutor. Currently we're using it like:
>
> synchronized(operator.getContainingTask.getCheckpointLock()) {
> // perform some operation
> }
>
> The purpose of getCheckpointLock in above code is to get the lock so that
> the operation can be performed synchronously without interference from any
> other checkpoint.
>
> Since MailboxExecutor is mainly used for asynchronization, I'm finding it
> difficult to convert the above piece of code to something that uses
> MailboxExecutor.
>
> I hope my problem makes sense, and I would really appreciate it if someone
> could help me to move forward with this problem.
>
> Thank you
>

Re: Dealing with incompatible state in Temporal Left Join on Schema Evolution (Table API)

[Martijn Visser]

Hi Michael,

A change in your schema would also result in a change in the columns that
you are selecting with your SELECT * query, would it not? Given that your
compiled plan contains the fields you're selecting, it would mean that your
compiled plan changes when you have a different schema. After all, if you
have a different schema, you would also need to alter the definition for
your table in the catalog.

I don't think we can and I also don't think we should try to fix just one
case: this should have a proper solution, something that has been brought
up in our conversation for Flink 2.0.

Best regards,

Martijn



On Fri, Jul 7, 2023 at 11:40 AM Michael Marino 
wrote:

> Hi Martijn,
>
> Thanks for the reply.
>
> I understand that this is generally the case for the Table/SQL API, though
> I didn't expect a change in the schema should effect a change in the
> compiled plan, especially since we are not changing the join keys, etc. I
> was wondering if there is a way to deal with this for this particular case
> by, e.g.:
>
> - doing something to enable a custom migration of the type
> - removing the state for the given operator when loading from savepoint
> - make the row data be "opaque" to the operator (e.g. keeping the data as
> *bytes* + the join key, and then doing a protobuf deserialization after the
> join).
> - something else entirely?
>
> We don't expect to be changing the input schema so often, but often enough
> (~ 1/ month) such that we need to be able to do this without data loss.
>
> If nothing is possible at the moment, my backup plan is to make use of our
> committed data (our final sink is an Iceberg table, also with the kinesis
> metadata) and restart the application using the latest info as our starting
> point. I'd like to avoid this, though, because it would certainly create a
> bit of complexity.
>
> Thanks,
> Mike
>
> On Fri, Jul 7, 2023 at 10:58 AM Martijn Visser 
> wrote:
>
>> Hi Michael,
>>
>> In the current Table API/SQL, there's no guarantee that a change to
>> either the query or the Flink version won't lead to state incompatibility.
>> That's also documented at
>> https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/ops/upgrading/#table-api--sql
>>
>> Best regards,
>>
>> Martijn
>>
>> On Fri, Jul 7, 2023 at 9:23 AM Michael Marino 
>> wrote:
>>
>>> Hi all,
>>>
>>> I had a question about how to allow a migration of state in a temporal
>>> left join. Let me first describe what we are doing:
>>>
>>> - We have an input stream (Kinesis stream) of device data and we are
>>> performing a left temporal join on this stream to enrich it with metadata
>>> about the device as the data arrives.
>>> - The Kinesis stream is watermarked using the arrival time, and the
>>> enrichment table (in this case a MySQL CDC connector from ververica) is
>>> watermarked with a constant, far-future timestamp (following the
>>> recommendation here; https://stackoverflow.com/a/69397954/533501), to
>>> ensure it doesn't stall and more or less immediately emits the row. The
>>> join is on the serial number, so:
>>>
>>> SELECT * FROM input_stream st
>>> LEFT JOIN enrichment_table FOR SYSTEM_TIME AS OF arrival_ts et
>>>   ON st.device_serial_number = et.sn;
>>>
>>> - We are using KDA, so Flink 1.15.2.
>>>
>>> Our issue is as follows. When we update the input schema of the Kinesis
>>> Stream (we are using the Kinesis Stream Table API connector, with protobuf
>>> format which I backported from 1.16), the state cannot be migrated since
>>> the new type is incompatible, e.g.:
>>>
>>>
>>> TemporalJoin[10] -> Calc[11] -> (... -> IcebergStreamWriter) (1/1)#10
>>> (...) switched from INITIALIZING to FAILED with failure cause:
>>> java.lang.RuntimeException: Error while getting state
>>> at
>>> org.apache.flink.runtime.state.DefaultKeyedStateStore.getMapState(DefaultKeyedStateStore.java:109)
>>> at
>>> org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getMapState(StreamingRuntimeContext.java:232)
>>> at
>>> org.apache.flink.table.runtime.operators.join.temporal.TemporalRowTimeJoinOperator.open(TemporalRowTimeJoinOperator.java:159)
>>> ...
>>> Caused by: org.apache.flink.util.StateMigrationException: The new state
>>> serializer
>>> (org.apache.flink.api.common.typeutils.base.MapSerializer@6621650b)
>>> must not be incompatible with the old state serializer
>>> (org.apache.flink.api.common.typeutils

Re: Dealing with incompatible state in Temporal Left Join on Schema Evolution (Table API)

[Martijn Visser]

Hi Michael,

In the current Table API/SQL, there's no guarantee that a change to either
the query or the Flink version won't lead to state incompatibility. That's
also documented at
https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/ops/upgrading/#table-api--sql

Best regards,

Martijn

On Fri, Jul 7, 2023 at 9:23 AM Michael Marino 
wrote:

> Hi all,
>
> I had a question about how to allow a migration of state in a temporal
> left join. Let me first describe what we are doing:
>
> - We have an input stream (Kinesis stream) of device data and we are
> performing a left temporal join on this stream to enrich it with metadata
> about the device as the data arrives.
> - The Kinesis stream is watermarked using the arrival time, and the
> enrichment table (in this case a MySQL CDC connector from ververica) is
> watermarked with a constant, far-future timestamp (following the
> recommendation here; https://stackoverflow.com/a/69397954/533501), to
> ensure it doesn't stall and more or less immediately emits the row. The
> join is on the serial number, so:
>
> SELECT * FROM input_stream st
> LEFT JOIN enrichment_table FOR SYSTEM_TIME AS OF arrival_ts et
>   ON st.device_serial_number = et.sn;
>
> - We are using KDA, so Flink 1.15.2.
>
> Our issue is as follows. When we update the input schema of the Kinesis
> Stream (we are using the Kinesis Stream Table API connector, with protobuf
> format which I backported from 1.16), the state cannot be migrated since
> the new type is incompatible, e.g.:
>
>
> TemporalJoin[10] -> Calc[11] -> (... -> IcebergStreamWriter) (1/1)#10
> (...) switched from INITIALIZING to FAILED with failure cause:
> java.lang.RuntimeException: Error while getting state
> at
> org.apache.flink.runtime.state.DefaultKeyedStateStore.getMapState(DefaultKeyedStateStore.java:109)
> at
> org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getMapState(StreamingRuntimeContext.java:232)
> at
> org.apache.flink.table.runtime.operators.join.temporal.TemporalRowTimeJoinOperator.open(TemporalRowTimeJoinOperator.java:159)
> ...
> Caused by: org.apache.flink.util.StateMigrationException: The new state
> serializer
> (org.apache.flink.api.common.typeutils.base.MapSerializer@6621650b) must
> not be incompatible with the old state serializer
> (org.apache.flink.api.common.typeutils.base.MapSerializer@97ed2886).
>
> In this particular case, the schema evolution was an added field, so
> something I would expect to be supported (or at least somehow supportable).
>
> Having a look at the code (
> https://github.com/apache/flink/blob/69e812688b43be9a0c4f79e6af81bc2d1d8a873e/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/temporal/TemporalRowTimeJoinOperator.java#L160),
> this is coming when the left state of the join is obtained, which, I
> suppose, is expected since the type there has changed.
>
> At the moment, the only way to fix this is to restart the application
> without state. This is of course, not ideal, since it can result in data
> loss. My question is if there is anyway around this? I'm not sure the state
> of the temporal join is actually required for our case because we really
> just need to ensure that all records from the upstream Kinesis are
> processed. It is not absolutely critical that the enrichment is
> "deterministic" or "exactly right at the point in time", it must just be
> eventually consistent, which is, e.g. why we use the far-future
> watermarking in the enrichment table. Here are some options I was
> considering, I'm not sure if they all make sense:
>
> - just remove the state for this operator on reboot somehow, and then
> start the application allowing non-restored state.
> - Somehow use a customer serializer to manage this.
> - If it is not possible to migrate the state at all, then we would somehow
> try to get the state of the Kinesis iterators (or maybe just a timestamp)
> and restart the job starting from that point in the Kinesis stream.
>
> I'm looking forward to your input, thanks for your help!
>
> Cheers,
> Mike
>
>
>
> --
>
> Michael Marino
>
> Principal Data Science & Analytics
>
> Phone:  +49 89 7167786 - 14
>
> linkedin.com/company/tadogmbh 
> | facebook.com/tado  | twitter.com/tado
>  | youtube.com/tado
> 
>
> www.tado.com | tado GmbH | Sapporobogen 6-8 | 80637 Munich | Germany
>
>  Managing Directors: Toon Bouten | Christian Deilmann | Johannes Schwarz |
> Josef Wenzl
>
> Registered with the Commercial Register Munich as HRB 194769 | VAT-No: DE
> 280012558
>

Re: Flink TCP socket custom source - savepoint cannot be taken

[Martijn Visser]

Hi Kamal,

There's no such limitation, so most likely this is related to the
implementation of your TCP source connector. Do keep in mind that just by
the nature of TCP, I doubt that you will have any guarantees when it comes
to this source. E.g. if you roll back to a savepoint of one day ago, how
will you be able to retrieve the data from your TCP source that's exactly
the same as it would be a day ago.

Best regards,

Martijn

On Sat, Jul 1, 2023 at 7:33 AM Kamal Mittal via user 
wrote:

> Hello Community,
>
>
>
> I have a requirement to read data coming over TCP socket stream and for
> the same written one custom source function reading data by TCP socket.
>
>
>
> Job is running successfully but while trying to take a savepoint, error
> comes that savepoint cannot be taken.
>
>
>
> Is there any limitation that TCP streams state can’t be taken (by
> checkpoint or savepoint) and re-stored later?
>
>
>
> Rgds,
>
> Kamal
>

Re: Data & Task distribution among the available Nodes

[Martijn Visser]

Hi Mahmoud,

While it's not an answer to your questions, I do want to point out
that the DataSet API is deprecated and will be removed in a future
version of Flink. I would recommend moving to either the Table API or
the DataStream API.

Best regards,

Martijn

On Thu, Jun 22, 2023 at 6:14 PM Mahmoud Awad  wrote:
>
> Hello everyone,
>
> I am trying to understand the mechanism by which Flink distributed the data 
> and the tasks among the nodes/task managers in the cluster, assuming all TMs 
> have equal resources. I am using the DataSet API on my own machine.
> I will try to address the issue with the following questions :
>
> -When we  firstly read the data from the source(Text,CSV..etc.), How does 
> Flink ensures the fairly distribution of data from the source to the next 
> subtask ?
>
> -Are there any preferences by which Flink will prefer a task manager on the 
> other(assuming all task managers have equal resources) ?
>
> - Based on what, will Flink choose to deploy a specific task in a specific 
> task manager ?
>
> I hope I was able to explain my point, thank you in advanced.
>
> Best regards
> Mahmoud
>
>
>
> Gesendet von Mail für Windows
>
>

[ANNOUNCE] Apache flink-connector-jdbc 3.1.1 released

[Martijn Visser]

The Apache Flink community is very happy to announce the release of
Apache flink-connector-jdbc v3.1.1. This version is compatible with
Flink 1.16 and Flink 1.17.

Apache Flink® is an open-source stream processing framework for
distributed, high-performing, always-available, and accurate data
streaming applications.

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

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

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

Regards,
Release Manager

Re: PyFlink SQL from Kafka to Iceberg issues

[Martijn Visser]

Hi Dani,

There are two things that I notice:

1. You're mixing different Flink versions (1.16 and 1.17): all Flink
artifacts should be from the same Flink version
2. S3 plugins need to be added to the plugins folder of Flink, because they
are loaded via the plugin mechanism. See
https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/filesystems/plugins/

Best regards,

Martijn

On Sat, Jun 24, 2023 at 1:22 PM Dániel Pálma  wrote:

> Hey folks,
>
> Nice to meet ya'll!
>
> I'm trying to get the following stack up and running locally:
>
> - Kafka as source
> - pyFlink SQL
> - Iceberg on top of MinIO
>
> The goal is to have a pyflink script that reads data from a Kafka topic,
> does some transformations, and dumps it into an iceberg table.
>
> I have everything, except for the pyflink app running in Docker, defined
> in a docker-compose.yml:
>
> version: "3.7"
> services:
>
> mariadb:
> image: 'mariadb:latest'
> hostname: mariadb
> container_name: mariadb
> ports:
> - '3306:3306'
> environment:
> MYSQL_ROOT_PASSWORD: admin
> MYSQL_USER: admin
> MYSQL_PASSWORD: admin
> MYSQL_DATABASE: metastore_db
> volumes:
> - ./mariadb-data:/var/lib/mysql
> networks:
> iceberg_net:
>
> hive-metastore:
> hostname: hive-metastore
> container_name: hive-metastore
> build:
> context: hive
> ports:
> - '9083:9083'
> environment:
> METASTORE_DB_HOSTNAME: mariadb
> depends_on:
> - mariadb
> networks:
> iceberg_net:
>
> minio:
> hostname: "minio"
> image: "minio/minio:latest"
> container_name: "minio"
> ports:
> - "9001:9001"
> - "9000:9000"
> command:
> - "server"
> - "/data"
> - "--console-address"
> - ":9001"
> volumes:
> - "minio:/data"
> environment:
> MINIO_ROOT_USER: "minio"
> MINIO_ROOT_PASSWORD: "minio123"
> networks:
> iceberg_net:
> aliases:
> - iceberg.minio
>
> mc:
> depends_on:
> - "minio"
> image: "minio/mc"
> container_name: "mc"
> entrypoint: >
> /bin/sh -c "
> until (/usr/bin/mc config host add minio http://minio:9000 minio
> minio123) do echo "...waiting..." && sleep 1; done;
> /usr/bin/mc rm -r --force minio/iceberg;
> /usr/bin/mc mb minio/iceberg;
> /usr/bin/mc policy set public minio/iceberg;
> tail -f /dev/null
> "
> networks:
> iceberg_net:
>
> broker:
> image: confluentinc/cp-kafka:7.4.0
> hostname: broker
> container_name: broker
> depends_on:
> - controller
> ports:
> - "9092:9092"
> - "9101:9101"
> environment:
> KAFKA_NODE_ID: 1
> KAFKA_LISTENER_SECURITY_PROTOCOL_MAP:
> 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
> KAFKA_ADVERTISED_LISTENERS:
> 'PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092'
> KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
> KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
> KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
> KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
> KAFKA_JMX_PORT: 9101
> KAFKA_JMX_HOSTNAME: localhost
> KAFKA_PROCESS_ROLES: 'broker'
> KAFKA_CONTROLLER_QUORUM_VOTERS: '2@controller:9093'
> KAFKA_LISTENERS: 'PLAINTEXT://broker:29092,PLAINTEXT_HOST://0.0.0.0:9092'
> KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
> KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
> KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'
> # Replace CLUSTER_ID with a unique base64 UUID using "bin/kafka-storage.sh
> random-uuid"
> # See
> https://docs.confluent.io/kafka/operations-tools/kafka-tools.html#kafka-storage-sh
> CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk'
> networks:
> iceberg_net:
>
> controller:
> image: confluentinc/cp-kafka:7.4.0
> hostname: controller
> container_name: controller
> ports:
> - "9093:9093"
> - "9102:9102"
> environment:
> KAFKA_NODE_ID: 2
> KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
> KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
> KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
> KAFKA_JMX_PORT: 9102
> KAFKA_JMX_HOSTNAME: localhost
> KAFKA_PROCESS_ROLES: 'controller'
> KAFKA_CONTROLLER_QUORUM_VOTERS: '2@controller:9093'
> KAFKA_LISTENERS: 'CONTROLLER://controller:9093'
> KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
> KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
> KAFKA_LOG_DIRS: '/tmp/kraft-controller-logs'
> # Replace CLUSTER_ID with a unique base64 UUID using "bin/kafka-storage.sh
> random-uuid"
> # See
> https://docs.confluent.io/kafka/operations-tools/kafka-tools.html#kafka-storage-sh
> CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk'
> networks:
> iceberg_net:
>
> schema-registry:
> image: confluentinc/cp-schema-registry:7.4.0
> hostname: schema-registry
> container_name: schema-registry
> depends_on:
> - broker
> - controller
> ports:
> - "8081:8081"
> environment:
> SCHEMA_REGISTRY_HOST_NAME: schema-registry
> SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'broker:29092'
> SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081
> networks:
> iceberg_net:
>
> control-center:
> image: confluentinc/cp-enterprise-control-center:7.4.0
> hostname: control-center
> container_name: control-center
> depends_on:
> - broker
> - controller
> - schema-registry
> ports:
> - "9021:9021"
> environment:
> CONTROL_CENTER_BOOTSTRAP_SERVERS: 'broker:29092'
> 

Re: Kafka source with idleness and alignment stops consuming

[Martijn Visser]

Hi Alexis,

There are a couple of recent Flink tickets on watermark alignment,
specifically https://issues.apache.org/jira/browse/FLINK-32414 and
https://issues.apache.org/jira/browse/FLINK-32420 - Could the later be also
applicable in your case?

Best regards,

Martijn

On Wed, Jun 28, 2023 at 11:33 AM Alexis Sarda-Espinosa <
sarda.espin...@gmail.com> wrote:

> Hello,
>
> just for completeness, I don't see the problem if I assign a different
> alignment group to each source, i.e. using only split-level watermark
> alignment.
>
> Regards,
> Alexis.
>
> Am Mi., 28. Juni 2023 um 08:13 Uhr schrieb haishui :
>
>> Hi,
>> I have the same trouble. This is really a bug.
>> `shouldWaitForAlignment` needs to be another change.
>>
>> By the way, a source will be marked as idle, when the source has waiting
>> for alignment for a long time. Is this a bug?
>>
>>
>>
>>
>>
>>
>> 在 2023-06-27 23:25:38，"Alexis Sarda-Espinosa" 
>> 写道：
>>
>> Hello,
>>
>> I am currently evaluating idleness and alignment with Flink 1.17.1 and
>> the externalized Kafka connector. My job has 3 sources whose watermark
>> strategies are defined like this:
>>
>> WatermarkStrategy.forBoundedOutOfOrderness(maxAllowedWatermarkDrift)
>> .withIdleness(idleTimeout)
>> .withWatermarkAlignment("group", maxAllowedWatermarkDrift,
>> Duration.ofSeconds(1L))
>>
>> The max allowed drift is currently 5 seconds, and my sources have an
>> idleTimeout of 1, 1.5, and 5 seconds.
>>
>> What I observe is that, when I restart the job, all sources publish
>> messages, but then 2 of them are marked as idle and never resume. I found
>> https://issues.apache.org/jira/browse/FLINK-31632, which should be fixed
>> in 1.17.1, but I don't think it's the same issue, my logs don't show
>> negative values:
>>
>> 2023-06-27 15:11:42,927 DEBUG
>> org.apache.flink.runtime.source.coordinator.SourceCoordinator:609 - New
>> reported watermark=Watermark @ 1687878696690 (2023-06-27 15:11:36.690) from
>> subTaskId=1
>> 2023-06-27 15:11:43,009 DEBUG
>> org.apache.flink.runtime.source.coordinator.SourceCoordinator:609 - New
>> reported watermark=Watermark @ 9223372036854775807 (292278994-08-17
>> 07:12:55.807) from subTaskId=0
>> 2023-06-27 15:11:43,091 DEBUG
>> org.apache.flink.runtime.source.coordinator.SourceCoordinator:609 - New
>> reported watermark=Watermark @ 9223372036854775807 (292278994-08-17
>> 07:12:55.807) from subTaskId=0
>> 2023-06-27 15:11:43,116 DEBUG
>> org.apache.flink.runtime.source.coordinator.SourceCoordinator:609 - New
>> reported watermark=Watermark @ 9223372036854775807 (292278994-08-17
>> 07:12:55.807) from subTaskId=0
>> 2023-06-27 15:11:43,298 INFO
>>  org.apache.flink.runtime.source.coordinator.SourceCoordinator:194 -
>> Distributing maxAllowedWatermark=1687878701690 to subTaskIds=[0, 1]
>> 2023-06-27 15:11:43,304 INFO
>>  org.apache.flink.runtime.source.coordinator.SourceCoordinator:194 -
>> Distributing maxAllowedWatermark=1687878701690 to subTaskIds=[0]
>> 2023-06-27 15:11:43,306 INFO
>>  org.apache.flink.runtime.source.coordinator.SourceCoordinator:194 -
>> Distributing maxAllowedWatermark=1687878701690 to subTaskIds=[0]
>> 2023-06-27 15:11:43,486 INFO
>>  org.apache.flink.runtime.source.coordinator.SourceCoordinator:194 -
>> Distributing maxAllowedWatermark=1687878701690 to subTaskIds=[]
>> 2023-06-27 15:11:43,489 INFO
>>  org.apache.flink.runtime.source.coordinator.SourceCoordinator:194 -
>> Distributing maxAllowedWatermark=1687878701690 to subTaskIds=[]
>> 2023-06-27 15:11:43,492 INFO
>>  org.apache.flink.runtime.source.coordinator.SourceCoordinator:194 -
>> Distributing maxAllowedWatermark=1687878701690 to subTaskIds=[]
>>
>> Does anyone know if I'm missing something or this is really a bug?
>>
>> Regards,
>> Alexis.
>>
>>

Re: Very long launch of the Flink application in BATCH mode

[Martijn Visser]

Hi Vladislav,

I think it might be worthwhile to upgrade to Flink 1.17, given the
improvements that have been made in Flink 1.16 and 1.17 on batch
processing. See for example the release notes of 1.17, with an entire
section on batch processing
https://flink.apache.org/2023/03/23/announcing-the-release-of-apache-flink-1.17/#batch-processing

Best regards,

Martijn

On Wed, Jun 28, 2023 at 7:27 PM Vladislav Keda 
wrote:

> Hi Shammon,
>
> When I set log.level=DEBUG I have no more logs except  *2023-06-21
> 14:51:30,921 DEBUG
> org.apache.flink.runtime.resourcemanager.active.ActiveResourceManager [] -
> Trigger heartbeat request.*
>
> Job freezes on stream graph generation. In STREAMING mode the job starts
> fast without same problems.
>
> ср, 28 июн. 2023 г. в 06:44, Shammon FY :
>
>> Hi Brendan,
>>
>> I think you may need to confirm which stage the job is blocked, the
>> client is submitting job or resourcemanage is scheduling job or tasks are
>> launching in TM? May be you need provide more information to help us to
>> figure the issue
>>
>> Best,
>> Shammon FY
>>
>> On Tuesday, June 27, 2023, Weihua Hu  wrote:
>>
>>> Hi, Brendan
>>>
>>> It looks like it's invoking your main method referring to the log. You
>>> can add more logs in the main method to figure out which part takes too
>>> long.
>>>
>>> Best,
>>> Weihua
>>>
>>>
>>> On Tue, Jun 27, 2023 at 5:06 AM Brendan Cortez <
>>> brendan.cortez...@gmail.com> wrote:
>>>
 No, I'm using a collection source + 20 same JDBC lookups + Kafka sink.

 On Mon, 26 Jun 2023 at 19:17, Yaroslav Tkachenko 
 wrote:

> Hey Brendan,
>
> Do you use a file source by any chance?
>
> On Mon, Jun 26, 2023 at 4:31 AM Brendan Cortez <
> brendan.cortez...@gmail.com> wrote:
>
>> Hi all!
>>
>> I'm trying to submit a Flink Job in Application Mode in the
>> Kubernetes cluster.
>>
>> I see some problems when an application has a big number of operators
>> (more than 20 same operators) - it freezes for ~6 minutes after
>> *2023-06-21 15:46:45,082 WARN
>>  org.apache.flink.connector.kafka.sink.KafkaSinkBuilder   [] - 
>> Property
>> [transaction.timeout.ms ] not specified.
>> Setting it to PT1H*
>>  and until
>>
>> *2023-06-21 15:53:20,002 INFO
>>  org.apache.flink.streaming.api.graph.StreamGraphGenerator[] - 
>> Disabled
>> Checkpointing. Checkpointing is not supported and not needed when 
>> executing
>> jobs in BATCH mode.*(logs in attachment)
>>
>> When I set log.level=DEBUG, I see only this message each 10 seconds:
>> *2023-06-21 14:51:30,921 DEBUG
>> org.apache.flink.runtime.resourcemanager.active.ActiveResourceManager [] 
>> -
>> Trigger heartbeat request.*
>>
>> Please, could you help me understand the cause of this problem and
>> how to fix it. I use the Flink 1.15.3 version.
>>
>> Thank you in advance!
>>
>> Best regards,
>> Brendan Cortez.
>>
>

Re: [Slack] Request to upload new invitation link

[Martijn Visser]

Thanks for reaching out Stephen. I've also updated the Slack invite link at
https://flink.apache.org/community/#slack

Best regards, Martijn

On Thu, Jun 29, 2023 at 3:20 AM yuxia  wrote:

> Hi, Stephen.
> Welcome to join Flink Slack channel. Here's my invitation link:
>
> https://join.slack.com/t/apache-flink/shared_invite/zt-1y7kmx7te-zUg1yfLdGu3Th9En_p4n~g
>
> Best regards,
> Yuxia
>
> --
> *发件人: *"Stephen Chu" 
> *收件人: *"User" 
> *抄送: *"Satyam Shanker" , "Vaibhav Gosain" <
> vaibhav.gos...@glean.com>, "Steve Jiang" 
> *发送时间: *星期四, 2023年 6 月 29日 上午 12:49:21
> *主题: *[Slack] Request to upload new invitation link
>
> Hi there,
> I'd love to join the Flink Slack channel, but it seems the link is
> outdated:
> https://join.slack.com/t/apache-flink/shared_invite/zt-1thin01ch-tYuj6Zwu8qf0QsivHY0anw
>
> Would someone be able to update or send me a new invite link?
>
> Thanks,
> Stephen
>
>

Re: [DISCUSS] Status of Statefun Project

[Martijn Visser via user]

Hi Galen,

Yes, I'll be more than happy to help with Statefun releases.

Best regards,

Martijn

On Tue, Jun 20, 2023 at 2:21 PM Galen Warren 
wrote:

> Thanks.
>
> Martijn, to answer your question, I'd need to do a small amount of work to
> get a PR ready, but not much. Happy to do it if we're deciding to restart
> Statefun releases -- are we?
>
> -- Galen
>
> On Sat, Jun 17, 2023 at 9:47 AM Tzu-Li (Gordon) Tai 
> wrote:
>
>> > Perhaps he could weigh in on whether the combination of automated tests
>> plus those smoke tests should be sufficient for testing with new Flink
>> versions
>>
>> What we usually did at the bare minimum for new StateFun releases was the
>> following:
>>
>>1. Build tests (including the smoke tests in the e2e module, which
>>covers important tests like exactly-once verification)
>>2. Updating the flink-statefun-playground repo and manually running
>>all language examples there.
>>
>> If upgrading Flink versions was the only change in the release, I'd
>> probably say that this is sufficient.
>>
>> Best,
>> Gordon
>>
>> On Thu, Jun 15, 2023 at 5:25 AM Martijn Visser 
>> wrote:
>>
>>> Let me know if you have a PR for a Flink update :)
>>>
>>> On Thu, Jun 8, 2023 at 5:52 PM Galen Warren via user <
>>> user@flink.apache.org> wrote:
>>>
>>>> Thanks Martijn.
>>>>
>>>> Personally, I'm already using a local fork of Statefun that is
>>>> compatible with Flink 1.16.x, so I wouldn't have any need for a released
>>>> version compatible with 1.15.x. I'd be happy to do the PRs to modify
>>>> Statefun to work with new versions of Flink as they come along.
>>>>
>>>> As for testing, Statefun does have unit tests and Gordon also sent me
>>>> instructions a while back for how to do some additional smoke tests which
>>>> are pretty straightforward. Perhaps he could weigh in on whether the
>>>> combination of automated tests plus those smoke tests should be sufficient
>>>> for testing with new Flink versions (I believe the answer is yes).
>>>>
>>>> -- Galen
>>>>
>>>>
>>>>
>>>> On Thu, Jun 8, 2023 at 8:01 AM Martijn Visser 
>>>> wrote:
>>>>
>>>>> Hi all,
>>>>>
>>>>> Apologies for the late reply.
>>>>>
>>>>> I'm willing to help out with merging requests in Statefun to keep them
>>>>> compatible with new Flink releases and create new releases. I do think
>>>>> that
>>>>> validation of the functionality of these releases depends a lot on
>>>>> those
>>>>> who do these compatibility updates, with PMC members helping out with
>>>>> the
>>>>> formal process.
>>>>>
>>>>> > Why can't the Apache Software Foundation allow community members to
>>>>> bring
>>>>> it up to date?
>>>>>
>>>>> There's nothing preventing anyone from reviewing any of the current
>>>>> PRs or
>>>>> opening new ones. However, none of them are approved [1], so there's
>>>>> also
>>>>> nothing to merge.
>>>>>
>>>>> > I believe that there are people and companies on this mailing list
>>>>> interested in supporting Apache Flink Stateful Functions.
>>>>>
>>>>> If so, then now is the time to show.
>>>>>
>>>>> Would there be a preference to create a release with Galen's merged
>>>>> compatibility update to Flink 1.15.2, or do we want to skip that and go
>>>>> straight to a newer version?
>>>>>
>>>>> Best regards,
>>>>>
>>>>> Martijn
>>>>>
>>>>> [1]
>>>>>
>>>>> https://github.com/apache/flink-statefun/pulls?q=is%3Apr+is%3Aopen+review%3Aapproved
>>>>>
>>>>> On Tue, Jun 6, 2023 at 3:55 PM Marco Villalobos <
>>>>> mvillalo...@kineteque.com>
>>>>> wrote:
>>>>>
>>>>> > Why can't the Apache Software Foundation allow community members to
>>>>> bring
>>>>> > it up to date?
>>>>> >
>>>>> > What's the process for that?
>>>>> >
>>>>> > I believe that there are people and companies on this mailing list
>>>>>

Re: [DISCUSS] Status of Statefun Project

[Martijn Visser]

Let me know if you have a PR for a Flink update :)

On Thu, Jun 8, 2023 at 5:52 PM Galen Warren via user 
wrote:

> Thanks Martijn.
>
> Personally, I'm already using a local fork of Statefun that is compatible
> with Flink 1.16.x, so I wouldn't have any need for a released version
> compatible with 1.15.x. I'd be happy to do the PRs to modify Statefun to
> work with new versions of Flink as they come along.
>
> As for testing, Statefun does have unit tests and Gordon also sent me
> instructions a while back for how to do some additional smoke tests which
> are pretty straightforward. Perhaps he could weigh in on whether the
> combination of automated tests plus those smoke tests should be sufficient
> for testing with new Flink versions (I believe the answer is yes).
>
> -- Galen
>
>
>
> On Thu, Jun 8, 2023 at 8:01 AM Martijn Visser 
> wrote:
>
>> Hi all,
>>
>> Apologies for the late reply.
>>
>> I'm willing to help out with merging requests in Statefun to keep them
>> compatible with new Flink releases and create new releases. I do think
>> that
>> validation of the functionality of these releases depends a lot on those
>> who do these compatibility updates, with PMC members helping out with the
>> formal process.
>>
>> > Why can't the Apache Software Foundation allow community members to
>> bring
>> it up to date?
>>
>> There's nothing preventing anyone from reviewing any of the current PRs or
>> opening new ones. However, none of them are approved [1], so there's also
>> nothing to merge.
>>
>> > I believe that there are people and companies on this mailing list
>> interested in supporting Apache Flink Stateful Functions.
>>
>> If so, then now is the time to show.
>>
>> Would there be a preference to create a release with Galen's merged
>> compatibility update to Flink 1.15.2, or do we want to skip that and go
>> straight to a newer version?
>>
>> Best regards,
>>
>> Martijn
>>
>> [1]
>>
>> https://github.com/apache/flink-statefun/pulls?q=is%3Apr+is%3Aopen+review%3Aapproved
>>
>> On Tue, Jun 6, 2023 at 3:55 PM Marco Villalobos <
>> mvillalo...@kineteque.com>
>> wrote:
>>
>> > Why can't the Apache Software Foundation allow community members to
>> bring
>> > it up to date?
>> >
>> > What's the process for that?
>> >
>> > I believe that there are people and companies on this mailing list
>> > interested in supporting Apache Flink Stateful Functions.
>> >
>> > You already had two people on this thread express interest.
>> >
>> > At the very least, we could keep the library versions up to date.
>> >
>> > There are only a small list of new features that might be worthwhile:
>> >
>> > 1. event time processing
>> > 2. state rest api
>> >
>> >
>> > On Jun 6, 2023, at 3:06 AM, Chesnay Schepler 
>> wrote:
>> >
>> > If you were to fork it *and want to redistribute it* then the short
>> > version is that
>> >
>> >1. you have to adhere to the Apache licensing requirements
>> >2. you have to make it clear that your fork does not belong to the
>> >Apache Flink project. (Trademarks and all that)
>> >
>> > Neither should be significant hurdles (there should also be plenty of
>> > online resources regarding 1), and if you do this then you can freely
>> share
>> > your fork with others.
>> >
>> > I've also pinged Martijn to take a look at this thread.
>> > To my knowledge the project hasn't decided anything yet.
>> >
>> > On 27/05/2023 04:05, Galen Warren wrote:
>> >
>> > Ok, I get it. No interest.
>> >
>> > If this project is being abandoned, I guess I'll work with my own fork.
>> Is
>> > there anything I should consider here? Can I share it with other people
>> who
>> > use this project?
>> >
>> > On Tue, May 16, 2023 at 10:50 AM Galen Warren 
>> 
>> > wrote:
>> >
>> >
>> > Hi Martijn, since you opened this discussion thread, I'm curious what
>> your
>> > thoughts are in light of the responses? Thanks.
>> >
>> > On Wed, Apr 19, 2023 at 1:21 PM Galen Warren 
>> 
>> > wrote:
>> >
>> >
>> > I use Apache Flink for stream processing, and StateFun as a hand-off
>> >
>> > point for the rest of the application.
>> > It serves well as a bridge between a Flink

Re: Async IO operator to write to DB

[Martijn Visser]

Hi Karthik,

In my opinion, it makes more sense to use a sink to leverage Scylla over
using Async IO. The primary use case for Async IO is enrichment, not for
writing to a sync.

Best regards,

Martijn

On Mon, Jun 12, 2023 at 4:10 PM Karthik Deivasigamani 
wrote:

> Thanks Martijn for your response.
> One thing I did not mention was that we are in the process of moving away
> from Cassandra to Scylla and would like to use the Scylla Java Driver for
> the following reason :
>
>> The Scylla Java driver is shard aware and contains extensions for a
>> tokenAwareHostPolicy. Using this policy, the driver can select a
>> connection to a particular shard based on the shard’s token. As a result,
>> latency is significantly reduced because there is no need to pass data
>> between the shards.
>>
> We were considering writing our own Sink to leverage Scylla Java Driver
> once the migration is done.
> ~
> Karthik
>
>
> On Mon, Jun 12, 2023 at 4:56 PM Martijn Visser 
> wrote:
>
>> Hi,
>>
>> Why wouldn't you just use the Flink Kafka connector and the Flink
>> Cassandra connector for your use case?
>>
>> Best regards,
>>
>> Martijn
>>
>> On Mon, Jun 12, 2023 at 12:03 PM Karthik Deivasigamani <
>> karthi...@gmail.com> wrote:
>>
>>> Hi,
>>>I have a use case where I need to read messages from a Kafka topic,
>>> parse it and write it to a database (Cassandra). Since Cassandra supports
>>> async APIs I was considering using Async IO operator for my writes. I do
>>> not need exactly-once semantics for my use-case.
>>> Is it okay to leverage the Async IO operator as a Sink (writing data
>>> into a DB)?
>>> ~
>>> Karthik
>>>
>>

Re: Flink source error handling

[Martijn Visser]

1. Like a regular Kafka client, so it depends on how you configure it.
2. Yes
3. It depends on the failure of course, but you could create something like
a Dead Letter Queue in case deserialization fails for your incoming message.

Best regards,

Martijn

On Sat, Jun 10, 2023 at 2:03 PM Anirban Dutta Gupta <
anir...@indicussoftware.com> wrote:

> Hello,
>
> Thanks for the guidance. We will surely think of moving to a newer version
> of Flink.
> Just a few followup questions when using KafkaSource..(sorry if I am being
> naive in my questions)
>
> 1. How does KafkaSource handle disruptions with the Kafka broker ? Will it
> keep on trying to connect and subscribe to the broker indefinitely or will
> it fail the Flink source after a certain number of retries ?
>
> 2. Will there be some log output in the Flink logs while the KafkaSource
> is trying again to connect to the broker after a disruption ?
>
> 3. In case of the source failing, is there a way in the Flink program
> using the KafkaSource to detect the error and add some error handling
> mechanism..for e.g. sending an alert mail to the stakeholders in case the
> source fails completely. (Something similar to
> "ActionRequestFailureHandler" for ElasticsearchSink)
>
> Many thanks in advance,
> Anirban
>
> On 09-06-2023 20:01, Martijn Visser wrote:
>
> Hi,
>
> This consumer should not be used. This only occurs in really old and no
> longer supported Flink versions. You should really upgrade to a newer
> version of Flink and use the KafkaSource.
>
> Best regards,
>
> Martijn
>
> On Fri, Jun 9, 2023 at 11:05 AM Anirban Dutta Gupta <
> anir...@indicussoftware.com> wrote:
>
>> Hello,
>>
>> We are using "FlinkKafkaConsumer011" as a Kafka source consumer for
>> Flink. Please guide on how to implement error handling mechanism for the
>> following:
>> 1. If the subscription to the Kafka topic gets lost, Kafka connection
>> gets disconnected.
>> In this case, is there any mechanism of re-subscribing to the Kafka
>> topic automatically in the program.
>>
>> 2. If there is any error in the FetchRecords of the consumer.
>>
>> Thanks and Regards,
>> Anirban
>>
>
>

Re: Async IO operator to write to DB

[Martijn Visser]

Hi,

Why wouldn't you just use the Flink Kafka connector and the Flink Cassandra
connector for your use case?

Best regards,

Martijn

On Mon, Jun 12, 2023 at 12:03 PM Karthik Deivasigamani 
wrote:

> Hi,
>I have a use case where I need to read messages from a Kafka topic,
> parse it and write it to a database (Cassandra). Since Cassandra supports
> async APIs I was considering using Async IO operator for my writes. I do
> not need exactly-once semantics for my use-case.
> Is it okay to leverage the Async IO operator as a Sink (writing data into
> a DB)?
> ~
> Karthik
>

Re: Flink source error handling

[Martijn Visser]

Hi,

This consumer should not be used. This only occurs in really old and no
longer supported Flink versions. You should really upgrade to a newer
version of Flink and use the KafkaSource.

Best regards,

Martijn

On Fri, Jun 9, 2023 at 11:05 AM Anirban Dutta Gupta <
anir...@indicussoftware.com> wrote:

> Hello,
>
> We are using "FlinkKafkaConsumer011" as a Kafka source consumer for
> Flink. Please guide on how to implement error handling mechanism for the
> following:
> 1. If the subscription to the Kafka topic gets lost, Kafka connection
> gets disconnected.
> In this case, is there any mechanism of re-subscribing to the Kafka
> topic automatically in the program.
>
> 2. If there is any error in the FetchRecords of the consumer.
>
> Thanks and Regards,
> Anirban
>

Re: [DISCUSS] Status of Statefun Project

[Martijn Visser]

Hi all,

Apologies for the late reply.

I'm willing to help out with merging requests in Statefun to keep them
compatible with new Flink releases and create new releases. I do think that
validation of the functionality of these releases depends a lot on those
who do these compatibility updates, with PMC members helping out with the
formal process.

> Why can't the Apache Software Foundation allow community members to bring
it up to date?

There's nothing preventing anyone from reviewing any of the current PRs or
opening new ones. However, none of them are approved [1], so there's also
nothing to merge.

> I believe that there are people and companies on this mailing list
interested in supporting Apache Flink Stateful Functions.

If so, then now is the time to show.

Would there be a preference to create a release with Galen's merged
compatibility update to Flink 1.15.2, or do we want to skip that and go
straight to a newer version?

Best regards,

Martijn

[1]
https://github.com/apache/flink-statefun/pulls?q=is%3Apr+is%3Aopen+review%3Aapproved

On Tue, Jun 6, 2023 at 3:55 PM Marco Villalobos 
wrote:

> Why can't the Apache Software Foundation allow community members to bring
> it up to date?
>
> What's the process for that?
>
> I believe that there are people and companies on this mailing list
> interested in supporting Apache Flink Stateful Functions.
>
> You already had two people on this thread express interest.
>
> At the very least, we could keep the library versions up to date.
>
> There are only a small list of new features that might be worthwhile:
>
> 1. event time processing
> 2. state rest api
>
>
> On Jun 6, 2023, at 3:06 AM, Chesnay Schepler  wrote:
>
> If you were to fork it *and want to redistribute it* then the short
> version is that
>
>1. you have to adhere to the Apache licensing requirements
>2. you have to make it clear that your fork does not belong to the
>Apache Flink project. (Trademarks and all that)
>
> Neither should be significant hurdles (there should also be plenty of
> online resources regarding 1), and if you do this then you can freely share
> your fork with others.
>
> I've also pinged Martijn to take a look at this thread.
> To my knowledge the project hasn't decided anything yet.
>
> On 27/05/2023 04:05, Galen Warren wrote:
>
> Ok, I get it. No interest.
>
> If this project is being abandoned, I guess I'll work with my own fork. Is
> there anything I should consider here? Can I share it with other people who
> use this project?
>
> On Tue, May 16, 2023 at 10:50 AM Galen Warren  
> 
> wrote:
>
>
> Hi Martijn, since you opened this discussion thread, I'm curious what your
> thoughts are in light of the responses? Thanks.
>
> On Wed, Apr 19, 2023 at 1:21 PM Galen Warren  
> 
> wrote:
>
>
> I use Apache Flink for stream processing, and StateFun as a hand-off
>
> point for the rest of the application.
> It serves well as a bridge between a Flink Streaming job and
> micro-services.
>
> This is essentially how I use it as well, and I would also be sad to see
> it sunsetted. It works well; I don't know that there is a lot of new
> development required, but if there are no new Statefun releases, then
> Statefun can only be used with older Flink versions.
>
> On Tue, Apr 18, 2023 at 10:04 PM Marco Villalobos  
> wrote:
>
>
> I am currently using Stateful Functions in my application.
>
> I use Apache Flink for stream processing, and StateFun as a hand-off
> point for the rest of the application.
> It serves well as a bridge between a Flink Streaming job and
> micro-services.
>
> I would be disappointed if StateFun was sunsetted.  Its a good idea.
>
> If there is anything I can do to help, as a contributor perhaps, please
> let me know.
>
>
> On Apr 3, 2023, at 2:02 AM, Martijn Visser  
> 
>
> wrote:
>
> Hi everyone,
>
> I want to open a discussion on the status of the Statefun Project [1]
>
> in Apache Flink. As you might have noticed, there hasn't been much
> development over the past months in the Statefun repository [2]. There is
> currently a lack of active contributors and committers who are able to help
> with the maintenance of the project.
>
> In order to improve the situation, we need to solve the lack of
>
> committers and the lack of contributors.
>
> On the lack of committers:
>
> 1. Ideally, there are some of the current Flink committers who have
>
> the bandwidth and can help with reviewing PRs and merging them.
>
> 2. If that's not an option, it could be a consideration that current
>
> committers only approve and review PRs, that are approved by those who are
> willing to contribute to Statefun and if the CI passe

Re: Raise alarm for corrupt records

[Martijn Visser]

Hi Kamal,

Documentation on the metrics can be found at
https://nightlies.apache.org/flink/flink-docs-master/docs/ops/metrics/

Best regards,

Martijn

On Wed, Jun 7, 2023 at 10:13 AM Kamal Mittal via user 
wrote:

> Hello,
>
>
>
> Thanks for quick reply.
>
>
>
> I am using parquet encoder/decoder and during decoding if any corrupt
> record comes then need to raise alarm and maintain metrices visible over
> Flink Metrices GUI.
>
>
>
> So any custom metrices can be created in Flink? Please give some reference
> of any such documentation.
>
>
>
> Rgds,
>
> Kamal
>
>
>
> *From:* Martijn Visser 
> *Sent:* 07 June 2023 12:31 PM
> *To:* Kamal Mittal 
> *Cc:* user@flink.apache.org
> *Subject:* Re: Raise alarm for corrupt records
>
>
>
> Hi Kamal,
>
>
>
> No, but it should be straightforward to create metrics or events for these
> types of situations and integrate them with your own alerting solution.
>
>
>
> Best regards,
>
>
>
> Martijn
>
>
>
> On Wed, Jun 7, 2023 at 8:25 AM Kamal Mittal via user <
> user@flink.apache.org> wrote:
>
> Hello Community,
>
>
>
> Is there any way Flink provides out of box to raise alarm for corrupt
> records (e.g. due to decoding failure) in between of running data pipeline
> and send this alarm to outside of task manager process?
>
>
>
> Rgds,
>
> Kamal
>
>

Re: Raise alarm for corrupt records

[Martijn Visser]

Hi Kamal,

No, but it should be straightforward to create metrics or events for these
types of situations and integrate them with your own alerting solution.

Best regards,

Martijn

On Wed, Jun 7, 2023 at 8:25 AM Kamal Mittal via user 
wrote:

> Hello Community,
>
>
>
> Is there any way Flink provides out of box to raise alarm for corrupt
> records (e.g. due to decoding failure) in between of running data pipeline
> and send this alarm to outside of task manager process?
>
>
>
> Rgds,
>
> Kamal
>

Re: Bulk storage of protobuf records in files

[Martijn Visser]

Hey Ryan,

I've never encountered a use case for writing Protobuf encoded files to a
filesystem.

Best regards,

Martijn

On Fri, May 26, 2023 at 6:39 PM Ryan Skraba via user 
wrote:

> Hello all!
>
> I discovered while investigating FLINK-32008[1] that we can write to the
> filesystem connector with the protobuf format, but today, the resulting
> file is pretty unlikely to be useful or rereadable.
>
> There's no real standard for storing many protobuf messages in a single
> file container, although the documentation mentions writing size-delimited
> messages sequentially[2].  In practice, I've never encountered protobuf
> binaries stored on filesystems without using some other sort of "framing"
> (like how parquet can be accessed with either an Avro or a protobuf
> oriented API).
>
> Does anyone have any use cases for bulk storage of protobuf messages on a
> filesystem?  Should these files just be considered temporary storage for
> Flink jobs, or do they need to be compatible with other systems?  Is there
> a splittable / compressable file format?
>
> The alternative might be to just forbid file storage for protobuf
> messages!  Any opinions?
>
> All my best, Ryan Skraba
>
> [1]: https://issues.apache.org/jira/browse/FLINK-32008
> [2]: https://protobuf.dev/programming-guides/techniques/#streaming
>

Re: flink14 batch mode can read.iceberg table but stream mode can not

[Martijn Visser]

Hi,

This question is better suited for the Iceberg community, since they've
built the Flink-Iceberg integration.

Best regards,

Martijn

On Wed, May 31, 2023 at 9:48 AM 湘晗刚 <1016465...@qq.com> wrote:

> flink14 batch mode can read iceberg table but stream mode can not ,why?
> Thanks in advance
> Kobe24
>

Re: Using pre-registered schemas with avro-confluent-registry format is not possible

[Martijn Visser]

Hi Jannik,

Can you share how you've set those properties, because I've been able to
use this without any problems.

Best regards,

Martijn

On Thu, Jun 1, 2023 at 2:43 PM Schmeier, Jannik 
wrote:

> Hello Thias,
>
>
>
> thank you for your answer.
>
>
>
> We've tested registering an existing (byte equal) schema a second time,
> but unfortunately the schema registry does still deny the request.
>
>
>
> Your last suggestion sounds promising, but I think there are some edge
> cases with this approach that will still cause an error. For example when
> writing to a new topic that’s empty, querying this topic before won't
> return any records and therefore the schema would not be put into the
> schemaRegistryClient cache.
>
>
>
> I'm still preferring a flag for the "avro-confluent-registry" format that
> will disable registering schemas and instead the format will just try to
> get the ID for a schema string from the registry. If there is an ID for
> that schema, Flink will use it. If there is none, an exception should be
> thrown.
>
> What do you think of that?
>
>
>
> Best regards,
>
> Jannik
>
>
>
>
>
> *Von:* Schwalbe Matthias 
> *Gesendet:* Mittwoch, 31. Mai 2023 13:33
> *An:* Schmeier, Jannik ; user@flink.apache.org
> *Betreff:* RE: Using pre-registered schemas with avro-confluent-registry
> format is not possible
>
>
>
>
>
> Hello Jannik,
>
>
>
> Some things to consider (I had a similar problem a couple of years before):
>
>- The schemaRegistryClient actually caches schema ids, so it will hit
>the schema registry only once,
>- The schema registered in schema registry needs to be byte-equal,
>otherwise schema registry considers it to be a new schema (version)
>- … to my best knowledge writing an existing schema to the schema
>registry does not fail because it is actually not written
>   - Could be that this is not entirely true as we had to replace the
>   whole schemaRegistryClient with our own implementation because the 
> existing
>   one could not be reconfigured to accept compressed answers from our r/o
>   proxy
>- if you manage to fill the cache of your schemaRegistryClient with
>the exact schema (e.g. by querying it beforehand) you might never run into
>the trouble
>
>
>
> Hope this helps … keep us posted 
>
>
>
> Thias
>
>
>
>
>
>
>
>
>
> *From:* Schmeier, Jannik 
> *Sent:* Wednesday, May 31, 2023 12:44 PM
> *To:* user@flink.apache.org
> *Subject:* Using pre-registered schemas with avro-confluent-registry
> format is not possible
>
>
>
> ⚠*EXTERNAL MESSAGE – **CAUTION: Think Before You Click *⚠
>
>
>
> Hello,
>
>
>
> I'm trying to use the avro-confluent-registry format with the Confluent
> Cloud Schema Registry in our company.
>
> Our schemas are managed via Terraform and global write access is denied
> for all Kafka clients in our environments (or at least in production).
>
> Therefore, when using the avro-confluent-registry format I'm getting an
> error when Flink is trying to serialize a row:
>
> java.lang.RuntimeException: Failed to serialize row.
>
> at
> org.apache.flink.formats.avro.AvroRowDataSerializationSchema.serialize(AvroRowDataSerializationSchema.java:90)
> ~[?:?]
>
> at
> org.apache.flink.formats.avro.AvroRowDataSerializationSchema.serialize(AvroRowDataSerializationSchema.java:40)
> ~[?:?]
>
> at
> org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaRecordSerializationSchema.serialize(DynamicKafkaRecordSerializationSchema.java:95)
> ~[?:?]
>
> at
> org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaRecordSerializationSchema.serialize(DynamicKafkaRecordSerializationSchema.java:36)
> ~[?:?]
>
> at
> org.apache.flink.connector.kafka.sink.KafkaWriter.write(KafkaWriter.java:196)
> ~[?:?]
>
> at
> org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperator.processElement(SinkWriterOperator.java:158)
> ~[flink-dist-1.17.0.jar:1.17.0]
>
> at
> org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:75)
> ~[flink-dist-1.17.0.jar:1.17.0]
>
> at
> org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50)
> ~[flink-dist-1.17.0.jar:1.17.0]
>
> at
> org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
> ~[flink-dist-1.17.0.jar:1.17.0]
>
> at
> org.apache.flink.table.runtime.util.StreamRecordCollector.collect(StreamRecordCollector.java:44)
> ~[flink-table-runtime-1.17.0.jar:1.17.0]
>
> at
> org.apache.flink.table.runtime.operators.sink.ConstraintEnforcer.processElement(ConstraintEnforcer.java:247)
> ~[flink-table-runtime-1.17.0.jar:1.17.0]
>
> at
> org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:75)
> 

Re: Using pre-registered schemas with avro-confluent-registry format is not possible

[Martijn Visser]

Hi Jannik,

By default, Kafka client applications automatically register new schemas
[1]. You should be able to influence that by using properties, e.g. setting:

'properties.auto.register.schemas' = 'false'
'properties.use.latest.version' = 'true'

Best regards,

Martijn

[1]
https://docs.confluent.io/platform/current/schema-registry/security/index.html#disabling-auto-schema-registration


On Wed, May 31, 2023 at 1:35 PM Schwalbe Matthias <
matthias.schwa...@viseca.ch> wrote:

>
>
> Hello Jannik,
>
>
>
> Some things to consider (I had a similar problem a couple of years before):
>
>- The schemaRegistryClient actually caches schema ids, so it will hit
>the schema registry only once,
>- The schema registered in schema registry needs to be byte-equal,
>otherwise schema registry considers it to be a new schema (version)
>- … to my best knowledge writing an existing schema to the schema
>registry does not fail because it is actually not written
>   - Could be that this is not entirely true as we had to replace the
>   whole schemaRegistryClient with our own implementation because the 
> existing
>   one could not be reconfigured to accept compressed answers from our r/o
>   proxy
>- if you manage to fill the cache of your schemaRegistryClient with
>the exact schema (e.g. by querying it beforehand) you might never run into
>the trouble
>
>
>
> Hope this helps … keep us posted 
>
>
>
> Thias
>
>
>
>
>
>
>
>
>
> *From:* Schmeier, Jannik 
> *Sent:* Wednesday, May 31, 2023 12:44 PM
> *To:* user@flink.apache.org
> *Subject:* Using pre-registered schemas with avro-confluent-registry
> format is not possible
>
>
>
> ⚠*EXTERNAL MESSAGE – **CAUTION: Think Before You Click *⚠
>
>
>
> Hello,
>
>
>
> I'm trying to use the avro-confluent-registry format with the Confluent
> Cloud Schema Registry in our company.
>
> Our schemas are managed via Terraform and global write access is denied
> for all Kafka clients in our environments (or at least in production).
>
> Therefore, when using the avro-confluent-registry format I'm getting an
> error when Flink is trying to serialize a row:
>
> java.lang.RuntimeException: Failed to serialize row.
>
> at
> org.apache.flink.formats.avro.AvroRowDataSerializationSchema.serialize(AvroRowDataSerializationSchema.java:90)
> ~[?:?]
>
> at
> org.apache.flink.formats.avro.AvroRowDataSerializationSchema.serialize(AvroRowDataSerializationSchema.java:40)
> ~[?:?]
>
> at
> org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaRecordSerializationSchema.serialize(DynamicKafkaRecordSerializationSchema.java:95)
> ~[?:?]
>
> at
> org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaRecordSerializationSchema.serialize(DynamicKafkaRecordSerializationSchema.java:36)
> ~[?:?]
>
> at
> org.apache.flink.connector.kafka.sink.KafkaWriter.write(KafkaWriter.java:196)
> ~[?:?]
>
> at
> org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperator.processElement(SinkWriterOperator.java:158)
> ~[flink-dist-1.17.0.jar:1.17.0]
>
> at
> org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:75)
> ~[flink-dist-1.17.0.jar:1.17.0]
>
> at
> org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50)
> ~[flink-dist-1.17.0.jar:1.17.0]
>
> at
> org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
> ~[flink-dist-1.17.0.jar:1.17.0]
>
> at
> org.apache.flink.table.runtime.util.StreamRecordCollector.collect(StreamRecordCollector.java:44)
> ~[flink-table-runtime-1.17.0.jar:1.17.0]
>
> at
> org.apache.flink.table.runtime.operators.sink.ConstraintEnforcer.processElement(ConstraintEnforcer.java:247)
> ~[flink-table-runtime-1.17.0.jar:1.17.0]
>
> at
> org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:75)
> ~[flink-dist-1.17.0.jar:1.17.0]
>
> at
> org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50)
> ~[flink-dist-1.17.0.jar:1.17.0]
>
> at
> org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
> ~[flink-dist-1.17.0.jar:1.17.0]
>
> at StreamExecCalc$2221.processElement_0_0(Unknown Source)
> ~[?:?]
>
> at
> StreamExecCalc$2221.processElement_0_0_rewriteGroup22_split310(Unknown
> Source) ~[?:?]
>
> at
> StreamExecCalc$2221.processElement_0_0_rewriteGroup22(Unknown Source) ~[?:?]
>
> at StreamExecCalc$2221.processElement_split308(Unknown
> Source) ~[?:?]
>
> at StreamExecCalc$2221.processElement(Unknown Source)
> ~[?:?]
>
> at
> 

Re: [ANNOUNCE] Apache Flink 1.17.1 released

[Martijn Visser]

Same here as with Flink 1.16.2, thank you Weijie and those who helped with
testing!

On Fri, May 26, 2023 at 1:08 PM weijie guo 
wrote:

>
> The Apache Flink community is very happy to announce the release of Apache 
> Flink 1.17.1, which is the first bugfix release for the Apache Flink 1.17 
> series.
>
>
>
>
> Apache Flink® is an open-source stream processing framework for distributed, 
> high-performing, always-available, and accurate data streaming applications.
>
>
>
> 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/news/2023/05/25/release-1.17.1.html
>
>
>
> The full release notes are available in Jira:
>
>
> https://issues.apache.org/jira/secure/ReleaseNote.jspa?projectId=12315522=12352886
>
>
>
>
> 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.1 released

[Martijn Visser]

Same here as with Flink 1.16.2, thank you Weijie and those who helped with
testing!

On Fri, May 26, 2023 at 1:08 PM weijie guo 
wrote:

>
> The Apache Flink community is very happy to announce the release of Apache 
> Flink 1.17.1, which is the first bugfix release for the Apache Flink 1.17 
> series.
>
>
>
>
> Apache Flink® is an open-source stream processing framework for distributed, 
> high-performing, always-available, and accurate data streaming applications.
>
>
>
> 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/news/2023/05/25/release-1.17.1.html
>
>
>
> The full release notes are available in Jira:
>
>
> https://issues.apache.org/jira/secure/ReleaseNote.jspa?projectId=12315522=12352886
>
>
>
>
> 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.16.2 released

[Martijn Visser]

Thank you Weijie and those who helped with testing!

On Fri, May 26, 2023 at 1:06 PM weijie guo 
wrote:

> The Apache Flink community is very happy to announce the release of
> Apache Flink 1.16.2, which is the second bugfix release for the Apache
> Flink 1.16 series.
>
>
>
> Apache Flink® is an open-source stream processing framework for
> distributed, high-performing, always-available, and accurate data
> streaming applications.
>
>
>
> 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/news/2023/05/25/release-1.16.2.html
>
>
>
> The full release notes are available in Jira:
>
>
> https://issues.apache.org/jira/secure/ReleaseNote.jspa?projectId=12315522=12352765
>
>
>
> 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.16.2 released

[Martijn Visser]

Thank you Weijie and those who helped with testing!

On Fri, May 26, 2023 at 1:06 PM weijie guo 
wrote:

> The Apache Flink community is very happy to announce the release of
> Apache Flink 1.16.2, which is the second bugfix release for the Apache
> Flink 1.16 series.
>
>
>
> Apache Flink® is an open-source stream processing framework for
> distributed, high-performing, always-available, and accurate data
> streaming applications.
>
>
>
> 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/news/2023/05/25/release-1.16.2.html
>
>
>
> The full release notes are available in Jira:
>
>
> https://issues.apache.org/jira/secure/ReleaseNote.jspa?projectId=12315522=12352765
>
>
>
> 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: Kafka Quotas & Consumer Group Client ID (Flink 1.15)

[Martijn Visser]

Hi Hatem,

Could it be that you don't have checkpointing enabled? Flink only commits
its offset when a checkpoint has been completed successfully, as explained
on
https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/kafka/#consumer-offset-committing

Best regards,

Martijn


On Tue, May 23, 2023 at 6:43 PM Hatem Mostafa  wrote:

> Hello,
>
> I have two questions that are related to each other:
>
> *First question:*
>
> I have been trying to set `client.id` to set a kafka client quota
>  for
> consumer_byte_rate since whenever our kafka job gets redeployed it reads a
> lot of data from our kafka cluster causing a denial of service for our
> kafka cluster. However `client.id` gets overridden by flink source here
> .
> How would I enforce quotas for flink kafka source?
>
> *Second question:*
>
> Also something I didn't quite understand when describing our consumer
> group in kafka why I don't see the metadata for the consumer group
> information (consumer id, client id & host) and I get that the consumer
> group has no active members but it's actually active and consuming.
>
> *Example describing a flink consumer group*
>
>> ./kafka-consumer-groups.sh --bootstrap-server kafka-server-address:9092
>>  --describe --group flink-consumer-group
>> Consumer group 'flink-consumer-group' has no active members.
>> GROUP   TOPIC   PARTITION  CURRENT-OFFSET
>>  LOG-END-OFFSET  LAG CONSUMER-ID HOSTCLIENT-ID
>> flink_consumer_group topic_name 1 514588965   514689721
>> 100756   -
>>- -
>
>
>
> *Example describing a normal consumer group written using a confluent
> kafka python library.*
>
>> ./kafka-consumer-groups.sh ---bootstrap-server kafka-server-address:9092
>> --describe --group python_confluent_kafka_consumer
>> GROUPTOPIC
>>  PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG CONSUMER-ID
>>
>>   HOST   CLIENT-ID
>> python_confluent_kafka_consumer topic_name   1
>>  17279532  17279908  376
>> python_confluent_kafka_consumer-345fa1d1-1f76-4e38-9aad-dcc120c5a52e
>> / python_confluent_kafka_consumer_client_id
>
>
>
> I am using flink version 1.15.
>
> Thanks,
> Hatem
>
>
>
>

Re: IRSA with Flink S3a connector

[Martijn Visser]

Hi Anuj,

I recalled another ticket on this topic, which had some things to test. I
don't know if that resolved the issue, can you verify it? See
https://issues.apache.org/jira/browse/FLINK-31095

Best regards,

Martijn

On Tue, May 23, 2023 at 7:04 AM Anuj Jain  wrote:

> Hello,
> Please provide some pointers on this issue.
>
> Thanks !!
>
> Regards
> Anuj
>
> On Fri, May 19, 2023 at 1:34 PM Anuj Jain  wrote:
>
>> Hi Community,
>> Looking forward to some advice on the problem.
>>
>> I also found this similar Jira, but not sure if a fix has been done for
>> the Hadoop connector - can someone confirm this.
>> [FLINK-23487] IRSA doesn't work with S3 - ASF JIRA (apache.org)
>> 
>>
>> Is there any other way to integrate Flink source/sink with AWS IAM from
>> EKS ?
>>
>> Regards
>> Anuj
>>
>> On Thu, May 18, 2023 at 12:41 PM Anuj Jain  wrote:
>>
>>> Hi,
>>> I have a flink job running on EKS, reading and writing data records to
>>> S3 buckets.
>>> I am trying to set up access credentials via AWS IAM.
>>> I followed this:
>>> https://docs.aws.amazon.com/eks/latest/userguide/iam-roles-for-service-accounts.html
>>>
>>> I have configured:
>>> com.amazonaws.auth.WebIdentityTokenCredentialsProvider as the credential
>>> provider in flink-conf.yaml for hadoop s3a connector, and annotated my
>>> service account with the role.
>>>
>>> When running the job, i am getting access denied error
>>> Exception:
>>> Caused by:
>>> com.amazonaws.services.securitytoken.model.AWSSecurityTokenServiceException:
>>> Not authorized to perform sts:AssumeRoleWithWebIdentity (Service:
>>> AWSSecurityTokenService; Status Code: 403; Error Code: AccessDenied;
>>> Request ID: 923df33a-802e-47e2-a203-0841aca03dd8; Proxy: null)
>>>
>>> I have tried to access S3 buckets from AWS CLI running in a pod with the
>>> same service account and that works.
>>>
>>> Am I using the correct credential provider for IAM integration, not sure
>>> if Hadoop S3a supports it.
>>> https://issues.apache.org/jira/browse/HADOOP-18154
>>>
>>> Please advise if I am doing anything wrong in setting up credentials via
>>> IAM.
>>>
>>> Regards
>>> Anuj Jain
>>>
>>

Re: Flink compatibility with JDK17

[Martijn Visser]

Hi Amenreet Singh Sodhi,

Flink is compatible with JDK8 and JDK11, not with JDK17. You can find the
Jira issue that tracks compatibility at
https://issues.apache.org/jira/browse/FLINK-15736. The biggest problem is
the Kryo serializer that's currently being used. That doesn't work with
JDK17, but upgrading would break compatibility so is not trivial.

Best regards,

Martijn

On Thu, May 11, 2023 at 3:08 PM amenreet sodhi  wrote:

> Hi,
>
> I wanted to confirm if Apache Flink is compatible with JDK17. If not, when
> can we expect this to happen? What major issues will I face if I try to
> compile it with JDK17?
>
> Thanks
> Regards
> Amenreet Singh Sodhi
>
>

[ANNOUNCE] Apache flink-connector-gcp-pubsub v3.0.1 released

[Martijn Visser]

The Apache Flink community is very happy to announce the release of Apache
flink-connector-gcp-pubsub v3.0.1. This release is compatible with Flink
1.16.x and Flink 1.17.x

Apache Flink® is an open-source stream processing framework for
distributed, high-performing, always-available, and accurate data streaming
applications.

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

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

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

Regards,
Release Manager

[ANNOUNCE] Apache flink-connector-gcp-pubsub v3.0.1 released

[Martijn Visser]

The Apache Flink community is very happy to announce the release of Apache
flink-connector-gcp-pubsub v3.0.1. This release is compatible with Flink
1.16.x and Flink 1.17.x

Apache Flink® is an open-source stream processing framework for
distributed, high-performing, always-available, and accurate data streaming
applications.

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

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

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

Regards,
Release Manager

[ANNOUNCE] Apache flink-connector-elasticsearch v3.0.1 released

[Martijn Visser]

The Apache Flink community is very happy to announce the release of Apache
flink-connector-elasticsearch v1.0.1. This release is compatible with Flink
1.16.x and Flink 1.17.x

Apache Flink® is an open-source stream processing framework for
distributed, high-performing, always-available, and accurate data streaming
applications.

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

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

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

Regards,
Release Manager

[ANNOUNCE] Apache flink-connector-elasticsearch v3.0.1 released

[Martijn Visser]

The Apache Flink community is very happy to announce the release of Apache
flink-connector-elasticsearch v1.0.1. This release is compatible with Flink
1.16.x and Flink 1.17.x

Apache Flink® is an open-source stream processing framework for
distributed, high-performing, always-available, and accurate data streaming
applications.

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

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

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

Regards,
Release Manager

[ANNOUNCE] Apache flink-connector-opensearch v1.0.1 released

[Martijn Visser]

The Apache Flink community is very happy to announce the release of Apache
flink-connector-opensearch v1.0.1. This release is compatible with Flink
1.16.x and Flink 1.17.x

Apache Flink® is an open-source stream processing framework for
distributed, high-performing, always-available, and accurate data streaming
applications.

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

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

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

Regards,
Release Manager

[ANNOUNCE] Apache flink-connector-opensearch v1.0.1 released

[Martijn Visser]

The Apache Flink community is very happy to announce the release of Apache
flink-connector-opensearch v1.0.1. This release is compatible with Flink
1.16.x and Flink 1.17.x

Apache Flink® is an open-source stream processing framework for
distributed, high-performing, always-available, and accurate data streaming
applications.

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

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

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

Regards,
Release Manager

[ANNOUNCE] Apache flink-connector-pulsar v4.0.0 released

[Martijn Visser]

The Apache Flink community is very happy to announce the release of Apache
flink-connector-pulsar v4.0.0. This release is compatible with Flink 1.17.x

Apache Flink® is an open-source stream processing framework for
distributed, high-performing, always-available, and accurate data streaming
applications.

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

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

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

Regards,
Release Manager

[ANNOUNCE] Apache flink-connector-pulsar v4.0.0 released

[Martijn Visser]

The Apache Flink community is very happy to announce the release of Apache
flink-connector-pulsar v4.0.0. This release is compatible with Flink 1.17.x

Apache Flink® is an open-source stream processing framework for
distributed, high-performing, always-available, and accurate data streaming
applications.

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

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

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

Regards,
Release Manager

[ANNOUNCE] Apache flink-shaded v17.0 released

[Martijn Visser]

The Apache Flink community is very happy to announce the release of Apache
flink-shaded v17.0.

Apache Flink® is an open-source stream processing framework for
distributed, high-performing, always-available, and accurate data streaming
applications.

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

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

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

Regards,
Release Manager

[ANNOUNCE] Apache flink-shaded v17.0 released

[Martijn Visser]

The Apache Flink community is very happy to announce the release of Apache
flink-shaded v17.0.

Apache Flink® is an open-source stream processing framework for
distributed, high-performing, always-available, and accurate data streaming
applications.

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

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

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

Regards,
Release Manager

[ANNOUNCE] Apache flink-connector-rabbitmq v3.0.1 released

[Martijn Visser]

The Apache Flink community is very happy to announce the release of Apache
flink-connector-rabbitmq v3.0.1. This release is compatible with Flink
1.16.x and Flink 1.17.x

Apache Flink® is an open-source stream processing framework for
distributed, high-performing, always-available, and accurate data streaming
applications.

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

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

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

Regards,
Release Manager

[ANNOUNCE] Apache flink-connector-rabbitmq v3.0.1 released

[Martijn Visser]

The Apache Flink community is very happy to announce the release of Apache
flink-connector-rabbitmq v3.0.1. This release is compatible with Flink
1.16.x and Flink 1.17.x

Apache Flink® is an open-source stream processing framework for
distributed, high-performing, always-available, and accurate data streaming
applications.

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

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

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

Regards,
Release Manager

Re: Encryption of parameters in flink-conf.yaml

[Martijn Visser]

Hi Anuj,

You can't provide the values for S3 in job code, since the S3 filesystems
are loaded via plugins. Credentials must be stored in flink-conf.yaml. The
recommended method for setting up credentials is by using IAM, not via
Access Keys. See
https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/filesystems/s3/#configure-access-credentials
for more details.

Best regards,

Martijn

On Tue, May 9, 2023 at 1:35 PM Anuj Jain  wrote:

> Hi,
>
> Thanks for the reply.
>
> Yes my flink deployment is on K8s but I am not using Flink-k8s operator.
> If i understood correctly, even with init-container the flink-conf.yaml
> (inside the container) would finally contain unencrypted values for access
> tokens. We don't want to persist such sensitive data unencrypted even
> inside running containers in files or config maps, due to some security
> constraints in my project.
> Can you please let me know if I missed something with the suggested
> solution.
>
> Problem with overriding configuration programmatically:
> When I removed the S3 properties from flink-conf.yaml and tried to provide
> it programmatically from the job code, the connection to S3 failed.
> I tried it with Application mode also on a standalone cluster but the
> behavior is the same.
>
> //My job main method (with default flink-conf.yaml):
> Configuration conf = new Configuration();
> conf.setString("s3a.access.key", );
> conf.setString("s3a.secret.key", );
> conf.setString("s3a.aws.credentials.provider",
> "org.apache.hadoop.fs.s3a.auth.AssumedRoleCredentialProvider");
> conf.setString("s3a.assumed.role.arn", );
> conf.setString("s3a.assumed.role.session.name", );
> conf.setString("s3a.assumed.role.session.duration", );
> conf.setString("s3a.assumed.role.sts.endpoint", );
> conf.setString("s3a.assumed.role.sts.endpoint.region", );
> final StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment(conf);
>
> // flink job program using DataStream
>
> env.execute("My job");
>
> With this i got connection exception
> Caused by: org.apache.flink.util.SerializedThrowable:
> com.amazonaws.SdkClientException: Unable to load AWS credentials from
> environment variables (AWS_ACCESS_KEY_ID (or AWS_ACCESS_KEY) and
> AWS_SECRET_KEY (or AWS_SECRET_ACCESS_KEY))
> at
> com.amazonaws.auth.EnvironmentVariableCredentialsProvider.getCredentials(EnvironmentVariableCredentialsProvider.java:50)
> ~[?:?]
> at
> org.apache.hadoop.fs.s3a.AWSCredentialProviderList.getCredentials(AWSCredentialProviderList.java:177)
> ~[?:?]
>
> When these values are given in flink-conf.yaml instead of job code, then
> connection was successful. Please guide if i am doing something incorrect
> w.r.t the job program.
>
> Regards
> Anuj
>
> On Mon, May 8, 2023 at 12:36 PM Biao Geng  wrote:
>
>> Hi Anuj,
>>
>> To my best knowledge, flink does not provide the encryption strategy
>> support for now. If you are using flink on k8s, it is possible to achieve
>> the encryption of parameters using the init container. You can check this
>> SO
>> 
>>  for
>> more detailed instructions.
>> Besides, it should be possible to override Configuration object in your
>> job code. Are you using Application mode to run the job?
>>
>> Best regards,
>> Biao Geng
>>
>> Anuj Jain  于2023年5月8日周一 13:55写道：
>>
>>> Hi Community,
>>> I am trying to create an amazon S3 filesystem distributor using flink
>>> and for this I am using hadoop S3a connector with Flink filesystem sink.
>>> My flink application would run in a non-AWS environment, on native
>>> cluster; so I need to put my access keys in flink configuration.
>>>
>>> For connecting to S3 storage, i am configuring flink-conf.yaml with the
>>> access credentials like
>>> s3.access.key: 
>>> s3.secret.key: 
>>> ... and some other parameters required for assuming AWS IAM role with
>>> s3a AssumedRoleCredentialProvider
>>>
>>> Is there a way to encrypt these parameters rather than putting them
>>> directly or is there any other way to supply them programmatically.
>>>
>>> I tried to set them programmatically using the Configuration object and
>>> supplying them with
>>> StreamExecutionEnvironment.getExecutionEnvironment(Configuration), in my
>>> job (rather than from flink-conf.yaml) but then the S3 connection failed. I
>>> think flink creates the connection pool at startup even before the job is
>>> started.
>>>
>>> Thanks and Regards
>>> Anuj Jain
>>>
>>

Re: [Discussion] - Release major Flink version to support JDK 17 (LTS)

[Martijn Visser]

Scala 2.12.7 doesn't compile on Java 17, see
https://issues.apache.org/jira/browse/FLINK-25000.

On Thu, Apr 27, 2023 at 3:11 PM Jing Ge  wrote:

> Thanks Tamir for the information. According to the latest comment of the
> task FLINK-24998, this bug should be gone while using the latest JDK 17. I
> was wondering whether it means that there are no more issues to stop us
> releasing a major Flink version to support Java 17? Did I miss something?
>
> Best regards,
> Jing
>
> On Thu, Apr 27, 2023 at 8:18 AM Tamir Sagi 
> wrote:
>
>> More details about the JDK bug here
>> https://bugs.openjdk.org/browse/JDK-8277529
>>
>> Related Jira ticket
>> https://issues.apache.org/jira/browse/FLINK-24998
>>
>> --
>> *From:* Jing Ge via user 
>> *Sent:* Monday, April 24, 2023 11:15 PM
>> *To:* Chesnay Schepler 
>> *Cc:* Piotr Nowojski ; Alexis Sarda-Espinosa <
>> sarda.espin...@gmail.com>; Martijn Visser ;
>> d...@flink.apache.org ; user 
>> *Subject:* Re: [Discussion] - Release major Flink version to support JDK
>> 17 (LTS)
>>
>>
>> *EXTERNAL EMAIL*
>>
>>
>> Thanks Chesnay for working on this. Would you like to share more info
>> about the JDK bug?
>>
>> Best regards,
>> Jing
>>
>> On Mon, Apr 24, 2023 at 11:39 AM Chesnay Schepler 
>> wrote:
>>
>> As it turns out Kryo isn't a blocker; we ran into a JDK bug.
>>
>> On 31/03/2023 08:57, Chesnay Schepler wrote:
>>
>>
>> https://github.com/EsotericSoftware/kryo/wiki/Migration-to-v5#migration-guide
>>
>> Kroy themselves state that v5 likely can't read v2 data.
>>
>> However, both versions can be on the classpath without classpath as v5
>> offers a versioned artifact that includes the version in the package.
>>
>> It probably wouldn't be difficult to migrate a savepoint to Kryo v5,
>> purely from a read/write perspective.
>>
>> The bigger question is how we expose this new Kryo version in the API. If
>> we stick to the versioned jar we need to either duplicate all current
>> Kryo-related APIs or find a better way to integrate other serialization
>> stacks.
>> On 30/03/2023 17:50, Piotr Nowojski wrote:
>>
>> Hey,
>>
>> > 1. The Flink community agrees that we upgrade Kryo to a later version,
>> which means breaking all checkpoint/savepoint compatibility and releasing a
>> Flink 2.0 with Java 17 support added and Java 8 and Flink Scala API support
>> dropped. This is probably the quickest way, but would still mean that we
>> expose Kryo in the Flink APIs, which is the main reason why we haven't been
>> able to upgrade Kryo at all.
>>
>> This sounds pretty bad to me.
>>
>> Has anyone looked into what it would take to provide a smooth migration
>> from Kryo2 -> Kryo5?
>>
>> Best,
>> Piotrek
>>
>> czw., 30 mar 2023 o 16:54 Alexis Sarda-Espinosa 
>> napisał(a):
>>
>> Hi Martijn,
>>
>> just to be sure, if all state-related classes use a POJO serializer, Kryo
>> will never come into play, right? Given FLINK-16686 [1], I wonder how many
>> users actually have jobs with Kryo and RocksDB, but even if there aren't
>> many, that still leaves those who don't use RocksDB for
>> checkpoints/savepoints.
>>
>> If Kryo were to stay in the Flink APIs in v1.X, is it impossible to let
>> users choose between v2/v5 jars by separating them like log4j2 jars?
>>
>> [1] https://issues.apache.org/jira/browse/FLINK-16686
>>
>> Regards,
>> Alexis.
>>
>> Am Do., 30. März 2023 um 14:26 Uhr schrieb Martijn Visser <
>> martijnvis...@apache.org>:
>>
>> Hi all,
>>
>> I also saw a thread on this topic from Clayton Wohl [1] on this topic,
>> which I'm including in this discussion thread to avoid that it gets lost.
>>
>> From my perspective, there's two main ways to get to Java 17:
>>
>> 1. The Flink community agrees that we upgrade Kryo to a later version,
>> which means breaking all checkpoint/savepoint compatibility and releasing a
>> Flink 2.0 with Java 17 support added and Java 8 and Flink Scala API support
>> dropped. This is probably the quickest way, but would still mean that we
>> expose Kryo in the Flink APIs, which is the main reason why we haven't been
>> able to upgrade Kryo at all.
>> 2. There's a contributor who makes a contribution that bumps Kryo, but
>> either a) automagically reads in all old checkpoints/savepoints in using
>> Kryo v2 and writes them to new snapshots using Kryo v5 (like is mentioned

Re: Moving from flinkkafkaconsumer to kafkasource issues

[Martijn Visser]

Hi,

Have you followed the documentation at
https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/kafka/#security
?

Best regards,

Martijn

On Fri, Apr 21, 2023 at 3:00 AM Shammon FY  wrote:

> Hi Naga
>
> Could you provide detailed error information? I think it may be useful for
> positioning the issue
>
> Best,
> Shammon FY
>
>
> On Fri, Apr 21, 2023 at 12:56 AM naga sudhakar 
> wrote:
>
>> Hi Team,
>> Greetings of the day..
>> we are on flink 1.16.1 version and using flinkkafkaconsumer today.
>> When I replaced it with kafkasource,it's failing with not able to connect
>> with kafka jaas configuration. Error says Kafka client entry not found in
>> /tmp/jass config file. We are passing the flonk runtime arg for the
>> security.auth.login conf details.  Same was working g with flink Kafka
>> consumer without any issues.
>> Did anyone  face this issue? Any suggestion? One of the stack over flow
>> suggestion was to have the content  of jaas config into java code,but not
>> use full
>>
>>
>>
>> Thanks,
>> Nagasudhakar
>>
>

Re: SplitEnumerator and SourceReader

[Martijn Visser]

Hi Kirti Dhar,

1. The SourceReader downloads the file, which is assigned to him by the
SplitEnumerator
2. This depends on the format; a BulkFormat like Parquet or ORC can be read
in batches of records at a time.
3. The SplitEnumerator runs on the JobManager, not on a TaskManager. Have
you read something different in the documentation?
4. Yes, because the enumerator needs to remember the paths of all currently
processed files. Depending on the use case, that can grow to be big. This
is documented at
https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/filesystem/#current-limitations

Best regards,

Martijn



On Thu, Apr 20, 2023 at 2:30 PM Kirti Dhar Upadhyay K via user <
user@flink.apache.org> wrote:

> Hi Community,
>
>
>
> I have started using file source of Flink 1.17.x recently.
>
> I was going through the FLIP-27 documentation and as much I understand
> SplitEnumerator lists files (splits) and assigns to SourceReader. A single
> instance of SplitEnumerator  runs whereas parallelism can be done on
> SourceReader side. I have below queries on same:
>
>
>
>1. Who actually downloads the file (let’s say the file is on S3)? Is
>it SplitEnumerator which downloads the files and then assign the splits to
>SourceReaders OR it only lists and give the path of file in split to
>SourceReader, which downloads the file and process?
>
>
>
>1. Is the complete file downloaded in one go? OR chunked downloading
>is also possible?
>
>
>
>1. I got that SplitEnumerator can be run on JobManager OR on single
>instance of TaskManager. How a user can configure it where to run?
>
>
>
>1. Is there any memory footprint impact if FileSource is running in
>streaming mode (continuous streaming)?
>
>
>
> Thanks for any help!
>
>
>
> Regards,
>
> Kirti Dhar
>

Re: Opensearch connector and oauth

[Martijn Visser]

Hi Michael,

I'm looping in Andrey since he has worked a lot on the Opensearch
connector. A contribution is very welcome in case this can be improved.

Best regards,

Martijn



On Tue, Apr 18, 2023 at 8:45 AM Michael Hempel Jørgensen 
wrote:

> Hi,
>
> we need to use OAuth2 (Client Credentials Flow) in Flink to authenticate
> and authorise against different services, initially Kafka and Opensearch.
> We have it working with Kafka using however it doesn't seem to be possible
> with the Opensearch Flink Connector (
> https://github.com/apache/flink-connector-opensearch).
>
> It looks like the connector only supports basic auth (user/pass) at the
> moment, and doesn't have an option for manually setting custom headers
> either. Is this correct?
>
> Is there any work in progress towards making this possible?
>
> We are willing to help contribute to code/discussions if necessary to get
> this implemented somehow.
>
> Regards,
> Michael
>
> --
> Michael Hempel-Jørgensen
> Software Pilot
> www.trifork.com
>
>
>

Re: Class not found exception for loaded class

[Martijn Visser]

Hi,

Only the S3 Presto and S3 Hadoop filesystem plugins don't rely on Hadoop
dependencies, all other filesystems do. See
https://nightlies.apache.org/flink/flink-docs-master/docs/dev/configuration/advanced/#hadoop-dependencies
for how to make them available.

Best regards,

Martijn

On Mon, Apr 17, 2023 at 2:04 PM Ruibin Xing  wrote:

> Hi community,
>
> I was testing Flink 1.17 on Kubernetes and ran into a strange class
> loading problem. In short, the logs
> show org.apache.hadoop.security.JniBasedUnixGroupsMappingWithFallback was
> loaded, however the program will throw ClassNotFoundException anyway.
>
> The exception was thrown by Aliyun OSS Filesystem plugin lib. the log
> shows:
>
> 2023-04-17 11:29:54.269 INFO
>  org.apache.flink.runtime.entrypoint.ClusterEntrypoint[] - Shutting
> KubernetesApplicationClusterEntrypoint down with application status FAILED.
> Diagnostics org.apache.flink.util.FlinkException: Could not create the ha
> services from the instantiated HighAvailabilityServicesFactory>
> at
> org.apache.flink.runtime.highavailability.HighAvailabilityServicesUtils.createCustomHAServices(HighAvailabilityServicesUtils.java:299)
> at
> org.apache.flink.runtime.highavailability.HighAvailabilityServicesUtils.createCustomHAServices(HighAvailabilityServicesUtils.java:285)
> at
> org.apache.flink.runtime.highavailability.HighAvailabilityServicesUtils.createHighAvailabilityServices(HighAvailabilityServicesUtils.java:145)
> at
> org.apache.flink.runtime.entrypoint.ClusterEntrypoint.createHaServices(ClusterEntrypoint.java:439)
> at
> org.apache.flink.runtime.entrypoint.ClusterEntrypoint.initializeServices(ClusterEntrypoint.java:382)
> at
> org.apache.flink.runtime.entrypoint.ClusterEntrypoint.runCluster(ClusterEntrypoint.java:282)
> at
> org.apache.flink.runtime.entrypoint.ClusterEntrypoint.lambda$startCluster$1(ClusterEntrypoint.java:232)
> at java.security.AccessController.doPrivileged(Native Method)
> at javax.security.auth.Subject.doAs(Subject.java:422)
> at
> org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1836)
> at
> org.apache.flink.runtime.security.contexts.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41)
> at
> org.apache.flink.runtime.entrypoint.ClusterEntrypoint.startCluster(ClusterEntrypoint.java:229)
> at
> org.apache.flink.runtime.entrypoint.ClusterEntrypoint.runClusterEntrypoint(ClusterEntrypoint.java:729)
> at
> org.apache.flink.kubernetes.entrypoint.KubernetesApplicationClusterEntrypoint.main(KubernetesApplicationClusterEntrypoint.java:86)
> Caused by: java.io.IOException: Could not create FileSystem for highly
> available storage path
> (oss://octopus-flink-test/checkpoints/ha/state-machine-test)
> at
> org.apache.flink.runtime.blob.BlobUtils.createFileSystemBlobStore(BlobUtils.java:102)
> at
> org.apache.flink.runtime.blob.BlobUtils.createBlobStoreFromConfig(BlobUtils.java:86)
> at
> org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory.createHAServices(KubernetesHaServicesFactory.java:41)
> at
> org.apache.flink.runtime.highavailability.HighAvailabilityServicesUtils.createCustomHAServices(HighAvailabilityServicesUtils.java:296)
> ... 13 more
> Caused by: java.lang.RuntimeException: java.lang.RuntimeException:
> java.lang.ClassNotFoundException: Class
> org.apache.hadoop.security.JniBasedUnixGroupsMappingWithFallback not found
> at
> org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.conf.Configuration.getClass(Configuration.java:2720)
> at
> org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.security.Groups.(Groups.java:107)
> at
> org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.security.Groups.(Groups.java:102)
> at
> org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.security.Groups.getUserToGroupsMappingService(Groups.java:451)
> at
> org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.security.UserGroupInformation.initialize(UserGroupInformation.java:338)
> at
> org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.security.UserGroupInformation.ensureInitialized(UserGroupInformation.java:300)
> at
> org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.security.UserGroupInformation.getCurrentUser(UserGroupInformation.java:575)
> at
> org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem.initialize(AliyunOSSFileSystem.java:341)
> at
> org.apache.flink.fs.osshadoop.OSSFileSystemFactory.create(OSSFileSystemFactory.java:103)
> at
> org.apache.flink.core.fs.PluginFileSystemFactory.create(PluginFileSystemFactory.java:62)
> at
> org.apache.flink.core.fs.FileSystem.getUnguardedFileSystem(FileSystem.java:508)
> at org.apache.flink.core.fs.FileSystem.get(FileSystem.java:409)
> at 

Re: [Discussion] - Take findify/flink-scala-api under Flink umbrella

[Martijn Visser]

Hi Alexey,

I would argue that it's not a problem from Flink's source code, the problem
was that Scala introduced a binary incompatible change in Scala 2.12.8. If
Flink wanted to allow an upgrade, it would mean breaking snapshot
compatibility. That's why Flink is still bound to be used with Scala
2.12.7. Any user can still decide to use a newer version of Scala, by
compiling Flink with a newer Scala version.

Given that Akka and Spark are predominantly built in Scala, I don't think
they are comparable with Flink, being a Java-first application. I still
would have expected that if the Scala type system and object serialization
in Flink were a problem for the users, they would have contributed more on
the Scala wrapper.

The code example's readability ultimately becomes a matter of personal
preference imho. I don't think that this is an argument we should use in
the discussion.

I would +1 Chesnay's idea to fork the Findify project first under
flink-extended and have volunteers step up there. It makes it possible to
mature the wrappers and see how it develops and gets used in the future.

Best regards,

Martijn

On Mon, Apr 17, 2023 at 10:19 AM Alexey Novakov 
wrote:

> Hi Martijn,
>
> Thanks for your reply and attention.
>
> 1. As I read Nick's report here
> https://issues.apache.org/jira/browse/FLINK-13414?focusedCommentId=17257763=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#comment-17257763
> Scala maintainers were blocked by Flink's source code inability to migrate
> from Scala 2.11 to newer versions easily. One strong reason is extensive
> Scala Macros usage in Flink Scala API, so that eventually few other Scala
> users developed 3-rd party Flink Wrappers on top of Java API once it became
> possible.
>
> 2. Scala wrapper is still needed due to the Scala type system and object
> serialization in Flink. You can not easily searilie Scala product type by
> ONLY using Java API. Scala collection types also differ from standard Java
> collections. If that would not be needed, I of course would not even start
> this discussion and continue to use Java API from Scala. Same principles of
> Scala and Java classes separation you can find in Akka and Apache Spark
> code bases.
>
> 3. Another point I did not mention in the first email, the Scala code
> examples look much more readable in Flink docs thanks to concise language
> syntax. It would be very helpful to keep them in Flink and make sure they
> work with Scala 2.13. and Scala 3. We would need to make sure if a user
> uses Scala code example from Flink docs, it works with Scala latest version
> without any issue. Otherwise, Scala users will have issues if they won't
> use an extra Scala wrapper for Java API. If that Scala wrapper is not an
> official part of Flink project, then it will be unsafe to use Scala at all.
> Günter has mentioned about it in his reply as well.
>
> Best regards,
> Alexey
>
> On Mon, Apr 17, 2023 at 9:27 AM Martijn Visser 
> wrote:
>
>> Hi Alexey,
>>
>> > Taking into account my Scala experience for the last 8 years, I predict
>> these wrappers will eventually be abandoned, unless such a Scala library is
>> a part of some bigger community like ASF.
>>
>> For the past couple of years, there have been no maintainers for Scala in
>> the Flink community. It was one of the reasons to deprecate the Scala APIs.
>> Given that the wrappers don't seem to have taken off outside of Flink, why
>> would moving them under the AS resolve this?
>>
>> > Also, non-official Scala API will lead people to play safe and choose
>> Java API only, even if they did want that at the beginning.
>>
>> Why would that be a problem? Wouldn't the fact that there are no
>> maintainers for the Scala wrappers actually indicate that Scala users are
>> actually fine with using the Java APIs, because else there would have been
>> improvements made towards the Scala wrappers?
>>
>> Best regards,
>>
>> Martijn
>>
>> On Sun, Apr 16, 2023 at 11:47 AM David Morávek  wrote:
>>
>>> cc dev@f.a.o
>>>
>>> On Sun, Apr 16, 2023 at 11:42 AM David Morávek  wrote:
>>>
>>> > Hi Alexey,
>>> >
>>> > I'm a bit skeptical because, looking at the project, I see a couple of
>>> red
>>> > flags:
>>> >
>>> > - The project is inactive. The last release and commit are both from
>>> the
>>> > last May.
>>> > - The project has not been adapted for the last two Flink versions,
>>> which
>>> > signals a lack of users.
>>> > - All commits are by a single person, which could mean that there is no
>>> > community around the project.
>

Re: Certificate renewal for Flink Kafka connectors

[Martijn Visser]

Hi Prateek,

You will need to stop and restart your jobs with the new connector
configuration.

Best regards,

Martijn

On Thu, Apr 13, 2023 at 10:10 AM Prateek Kohli 
wrote:

> Hi,
>
> I am using Flink Kafka connectors to communicate with Kafka broker over
> mutual TLS.
> Is there any way or recommendation to handle certificate renewal for these
> Kafka clients.
>
> I am monitoring the pem files and recreating the keystore/truststore(jks)
> on renewal, but how can I reload these to already running Flink Kafka
> connectors?
>
> Flink version - 1.16
>
> Thanks!!
>

Re: [Discussion] - Take findify/flink-scala-api under Flink umbrella

[Martijn Visser]

Hi Alexey,

> Taking into account my Scala experience for the last 8 years, I predict
these wrappers will eventually be abandoned, unless such a Scala library is
a part of some bigger community like ASF.

For the past couple of years, there have been no maintainers for Scala in
the Flink community. It was one of the reasons to deprecate the Scala APIs.
Given that the wrappers don't seem to have taken off outside of Flink, why
would moving them under the AS resolve this?

> Also, non-official Scala API will lead people to play safe and choose
Java API only, even if they did want that at the beginning.

Why would that be a problem? Wouldn't the fact that there are no
maintainers for the Scala wrappers actually indicate that Scala users are
actually fine with using the Java APIs, because else there would have been
improvements made towards the Scala wrappers?

Best regards,

Martijn

On Sun, Apr 16, 2023 at 11:47 AM David Morávek  wrote:

> cc dev@f.a.o
>
> On Sun, Apr 16, 2023 at 11:42 AM David Morávek  wrote:
>
> > Hi Alexey,
> >
> > I'm a bit skeptical because, looking at the project, I see a couple of
> red
> > flags:
> >
> > - The project is inactive. The last release and commit are both from the
> > last May.
> > - The project has not been adapted for the last two Flink versions, which
> > signals a lack of users.
> > - All commits are by a single person, which could mean that there is no
> > community around the project.
> > - There was no external contribution (except the Scala bot).
> > - There is no fork of the project (except the Scala bot).
> >
> > >  As I know, FIndify does not want or cannot maintain this library.
> >
> > Who are the users of the library? I'd assume Findify no longer uses it if
> > they're abandoning it.
> >
> > > which would be similar to the StateFun
> >
> > We're currently dealing with a lack of maintainers for StateFun, so we
> > should have a solid building ground around the project to avoid the same
> > issue.
> >
> >
> > I think there is value in having a modern Scala API, but we should have a
> > bigger plan to address the future of Flink Scala APIs than importing an
> > unmaintained library and calling it a day. I suggest starting a thread on
> > the dev ML and concluding the overall plan first.
> >
> > Best,
> > D.
> >
> > On Sun, Apr 16, 2023 at 10:48 AM guenterh.lists <
> guenterh.li...@bluewin.ch>
> > wrote:
> >
> >> Hello Alexey
> >>
> >> Thank you for your initiative and your suggestion!
> >>
> >> I can only fully support the following statements in your email:
> >>
> >>  >Taking into account my Scala experience for the last 8 years, I
> >> predict these wrappers will eventually be abandoned, unless such a Scala
> >> library is a part of some bigger community like ASF.
> >>  >Also, non-official Scala API will lead people to play safe and choose
> >> Java API only, even if they didn't want that at the beginning.
> >>
> >> Second sentence is my current state.
> >>
> >>  From my point of view it would be very unfortunate if the Flink project
> >> would lose the Scala API and thus the integration of concise, flexible
> >> and future-oriented language constructs of the Scala language (and
> >> further development of version 3).
> >>
> >> Documentation of the API is essential. I would be interested to support
> >> this efforts.
> >>
> >> Best wishes
> >>
> >> Günter
> >>
> >>
> >> On 13.04.23 15:39, Alexey Novakov via user wrote:
> >> > Hello Flink PMCs and Flink Scala Users,
> >> >
> >> > I would like to propose an idea to take the 3rd party Scala API
> >> > findify/flink-scala-api 
> >> > project into the Apache Flink organization.
> >> >
> >> > *Motivation *
> >> >
> >> > The Scala-free Flink idea was finally implemented by the 1.15 release
> >> and
> >> > allowed Flink users to bring their own Scala version and use it via
> the
> >> > Flink Java API. See blog-post here: Scala Free in One Fifteen
> >> > .
> Also,
> >> > existing Flink Scala API will be deprecated, because it is too hard to
> >> > upgrade it to Scala 2.13 or 3.
> >> >
> >> > Taking into account my Scala experience for the last 8 years, I
> predict
> >> > these wrappers will eventually be abandoned, unless such a Scala
> >> library is
> >> > a part of some bigger community like ASF.
> >> > Also, non-official Scala API will lead people to play safe and choose
> >> Java
> >> > API only, even if they did want that at the beginning.
> >> >
> >> > https://github.com/findify/flink-scala-api has already advanced and
> >> > implemented Scala support for 2.13 and 3 versions on top of Flink Java
> >> API.
> >> > As I know, FIndify does not want or does not have a capacity to
> maintain
> >> > this library. I propose to fork this great library and create a new
> >> Flink
> >> > project with its own version and build process (SBT, not Maven), which
> >> > would be similar to the StateFun or FlinkML 

[DISCUSS] Status of Statefun Project

[Martijn Visser]

Hi everyone,

I want to open a discussion on the status of the Statefun Project [1] in
Apache Flink. As you might have noticed, there hasn't been much development
over the past months in the Statefun repository [2]. There is currently a
lack of active contributors and committers who are able to help with the
maintenance of the project.

In order to improve the situation, we need to solve the lack of committers
and the lack of contributors.

On the lack of committers:

1. Ideally, there are some of the current Flink committers who have the
bandwidth and can help with reviewing PRs and merging them.
2. If that's not an option, it could be a consideration that current
committers only approve and review PRs, that are approved by those who are
willing to contribute to Statefun and if the CI passes

On the lack of contributors:

3. Next to having this discussion on the Dev and User mailing list, we can
also create a blog with a call for new contributors on the Flink project
website, send out some tweets on the Flink / Statefun twitter accounts,
post messages on Slack etc. In that message, we would inform how those that
are interested in contributing can start and where they could reach out for
more information.

There's also option 4. where a group of interested people would split
Statefun from the Flink project and make it a separate top level project
under the Apache Flink umbrella (similar as recently has happened with
Flink Table Store, which has become Apache Paimon).

If we see no improvements in the coming period, we should consider
sunsetting Statefun and communicate that clearly to the users.

I'm looking forward to your thoughts.

Best regards,

Martijn

[1] https://nightlies.apache.org/flink/flink-statefun-docs-master/
[2] https://github.com/apache/flink-statefun

Re: flink 1.17 connector adapt

[Martijn Visser]

Hi Tian,

Thanks for flagging this. This is the first time that we've released a
Flink version with connectors externalized and we're still discussing
what's the best way to release connectors for new versions in a simple way.
This is something that we're trying to get done asap.

Best regards,

Martijn

On Fri, Mar 31, 2023 at 7:22 AM tian tian  wrote:

> Hi, Flink 1.17 has been released, but elasticsearch and rabbitmq have not
> been adapted to 1.17. Is there any plan to adapt it?
>

Re: [Discussion] - Release major Flink version to support JDK 17 (LTS)

[Martijn Visser]

Hi all,

I also saw a thread on this topic from Clayton Wohl [1] on this topic,
which I'm including in this discussion thread to avoid that it gets lost.

>From my perspective, there's two main ways to get to Java 17:

1. The Flink community agrees that we upgrade Kryo to a later version,
which means breaking all checkpoint/savepoint compatibility and releasing a
Flink 2.0 with Java 17 support added and Java 8 and Flink Scala API support
dropped. This is probably the quickest way, but would still mean that we
expose Kryo in the Flink APIs, which is the main reason why we haven't been
able to upgrade Kryo at all.
2. There's a contributor who makes a contribution that bumps Kryo, but
either a) automagically reads in all old checkpoints/savepoints in using
Kryo v2 and writes them to new snapshots using Kryo v5 (like is mentioned
in the Kryo migration guide [2][3] or b) provides an offline tool that
allows users that are interested in migrating their snapshots manually
before starting from a newer version. That potentially could prevent the
need to introduce a new Flink major version. In both scenarios, ideally the
contributor would also help with avoiding the exposure of Kryo so that we
will be in a better shape in the future.

It would be good to get the opinion of the community for either of these
two options, or potentially for another one that I haven't mentioned. If it
appears that there's an overall agreement on the direction, I would propose
that a FLIP gets created which describes the entire process.

Looking forward to the thoughts of others, including the Users (therefore
including the User ML).

Best regards,

Martijn

[1]  https://lists.apache.org/thread/qcw8wy9dv8szxx9bh49nz7jnth22p1v2
[2] https://lists.apache.org/thread/gv49jfkhmbshxdvzzozh017ntkst3sgq
[3] https://github.com/EsotericSoftware/kryo/wiki/Migration-to-v5

On Sun, Mar 19, 2023 at 8:16 AM Tamir Sagi 
wrote:

> I agree, there are several options to mitigate the migration from v2 to v5.
> yet, Oracle roadmap is to end JDK 11 support in September this year.
>
>
>
> 
> From: ConradJam 
> Sent: Thursday, March 16, 2023 4:36 AM
> To: d...@flink.apache.org 
> Subject: Re: [Discussion] - Release major Flink version to support JDK 17
> (LTS)
>
> EXTERNAL EMAIL
>
>
>
> Thanks for your start this discuss
>
>
> I have been tracking this problem for a long time, until I saw a
> conversation in ISSUSE a few days ago and learned that the Kryo version
> problem will affect the JDK17 compilation of snapshots [1] FLINK-24998 ,
>
> As @cherry said it ruined our whole effort towards JDK17
>
> I am in favor of providing an external tool to migrate from Kryo old
> version checkpoint to the new Kryo new checkpoint at one time (Maybe this
> tool start in flink 2.0 ?), does this tool currently have any plans or
> ideas worth discuss
>
>
> I think it should not be difficult to be compatible with JDK11 and JDK17.
> We should indeed abandon JDK8 in 2.0.0. It is also mentioned in the doc
> that it is marked as Deprecated [2]
>
>
> Here I add that we need to pay attention to the version of Scala and the
> version of JDK17
>
>
> [1] FLINK-24998  IGSEGV in Kryo / C2 CompilerThread on Java 17
> https://issues.apache.org/jira/browse/FLINK-24998
>
> [2] FLINK-30501 Update Flink build instruction to deprecate Java 8 instead
> of requiring Java 11  https://issues.apache.org/jira/browse/FLINK-30501
>
> Tamir Sagi  于2023年3月16日周四 00:54写道：
>
> > Hey dev community,
> >
> > I'm writing this email to kick off a discussion following this epic:
> > FLINK-15736.
> >
> > We are moving towards JDK 17 (LTS) , the only blocker now is Flink which
> > currently remains on JDK 11 (LTS). Flink does not support JDK 17 yet,
> with
> > no timeline,  the reason, based on the aforementioned ticket is the
> > following tickets
> >
> >   1.  FLINK-24998 - SIGSEGV in Kryo / C2 CompilerThread on Java 17<
> > https://issues.apache.org/jira/browse/FLINK-24998>.
> >   2.  FLINK-3154 - Update Kryo version from 2.24.0 to latest Kryo LTS
> > version
> >
> > My question is whether it is possible to release a major version (Flink
> > 2.0.0) using the latest Kryo version for those who don't need to restore
> > old savepoints/checkpoints in newer format.
> >
> >   1.  Leverage JDK 17 features within JVM
> >   2.  Moving from the old format to the newer one will be handled only
> > once - a mitigation can be achieved by a conversion tool or external
> > serializers, both can be provided later on.
> >
> > I'd like to emphasize that the next JDK LTS (21) will be released this
> > September.  furthermore, Flink already supports JDK 12-15, which is very
> > close to JDK 17 (LTS) - that was released in September 2021.  JDK 11 will
> > become a legacy soon, as more frameworks moving towards JDK 17 and are
> less
> > likely to support JDK 11 in the near future. (For example, Spring 

[ANNOUNCE] TAC supporting Berlin Buzzwords

[Martijn Visser]

Hi everyone,

I'm forwarding the following information from the ASF Travel Assistance
Committee (TAC):

---

Hi All,

The ASF Travel Assistance Committee is supporting taking up to six (6)
people
to attend Berlin Buzzwords [1] In June this year.

This includes Conference passes, and travel & accommodation as needed.

Please see our website at https://tac.apache.org for more information and
how to apply.

Applications close on 15th April.

Good luck to those that apply.

Gavin McDonald (VP TAC)

---

Best regards,

Martijn

[1] https://2023.berlinbuzzwords.de/

Re: Unexpected behaviour when configuring both `withWatermarkAlignment` & `withIdleness`

[Martijn Visser]

Hi Reem

My thinking is that this might be related to recently reported
https://issues.apache.org/jira/browse/FLINK-31632.

Best regards,

Martijn

On Wed, Mar 29, 2023 at 7:07 PM Reem Razak via user 
wrote:

> Hey Martijn,
>
> The version is 1.16.0
>
> On Wed, Mar 29, 2023 at 5:43 PM Martijn Visser 
> wrote:
>
>> Hi Reem,
>>
>> What's the Flink version where you're encountering this issue?
>>
>> Best regards,
>>
>> Martijn
>>
>> On Wed, Mar 29, 2023 at 5:18 PM Reem Razak via user <
>> user@flink.apache.org> wrote:
>>
>>> Hey there!
>>>
>>> We are seeing a second Flink pipeline encountering similar issues when
>>> configuring both `withWatermarkAlignment` and `withIdleness`. The
>>> unexpected behaviour gets triggered after a Kafka cluster failover. Any
>>> thoughts on there being an incompatibility between the two?
>>>
>>> Thanks!
>>>
>>> On Wed, Nov 9, 2022 at 6:42 PM Reem Razak 
>>> wrote:
>>>
>>>> Hi there,
>>>>
>>>> We are integrating the watermark alignment feature into a pipeline with
>>>> a Kafka source during a "backfill"- i.e. playing from an earlier Kafka
>>>> offset. While testing, we noticed some unexpected behaviour in the
>>>> watermark advancement which was resolved by removing `withIdleness` from
>>>> our watermark strategy.
>>>>
>>>>
>>>> val watermarkStrategy = WatermarkStrategy
>>>>   .forBoundedOutOfOrderness(Duration.ofMinutes(1))
>>>>   .withTimestampAssigner(new
>>>> TimestampedEventTimestampAssigner[Event])
>>>>   .withWatermarkAlignment("alignment-group-1",
>>>> Duration.ofMinutes(1))
>>>>   .withIdleness(Duration.ofMinutes(5))
>>>>
>>>> I have attached a couple of screenshots of the watermarkAlignmentDrift
>>>> metric. As you can see, the behaviour seems normal until a sudden drop in
>>>> the value to ~ Long.MIN_VALUE, which causes the pipeline to stop emitting
>>>> records completely from the source. Furthermore, the logs originating from
>>>> from
>>>> https://github.com/dawidwys/flink/blob/888162083fbb5f62f809e5270ad968db58cc9c5b/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinator.java#L174
>>>> also indicate that the `maxAllowedWatermark` switches to ~ Long.MIN_VALUE.
>>>>
>>>> We found that modifying the `updateInterval` passed into the alignment
>>>> parameters seemed to correlate with how long the pipeline would operate
>>>> before stopping - a larger interval of 20 minutes would encounter the issue
>>>> later than an interval of 1 second.
>>>>
>>>> We are wondering if a bug exists when using both `withIdleness` and
>>>> `withWatermarkAlignment`. Might it be related to
>>>> https://issues.apache.org/jira/browse/FLINK-28975, or is there
>>>> possibly a race condition in the watermark emission? We do not necessarily
>>>> need to have both configured at the same time, but we were also surprised
>>>> by the behaviour of the application. Has anyone run into a similar issue or
>>>> have further insight?
>>>>
>>>> Much Appreciated,
>>>> - Reem
>>>>
>>>>
>>>>
>>>>

Re: Unexpected behaviour when configuring both `withWatermarkAlignment` & `withIdleness`

[Martijn Visser]

Hi Reem,

What's the Flink version where you're encountering this issue?

Best regards,

Martijn

On Wed, Mar 29, 2023 at 5:18 PM Reem Razak via user 
wrote:

> Hey there!
>
> We are seeing a second Flink pipeline encountering similar issues when
> configuring both `withWatermarkAlignment` and `withIdleness`. The
> unexpected behaviour gets triggered after a Kafka cluster failover. Any
> thoughts on there being an incompatibility between the two?
>
> Thanks!
>
> On Wed, Nov 9, 2022 at 6:42 PM Reem Razak  wrote:
>
>> Hi there,
>>
>> We are integrating the watermark alignment feature into a pipeline with a
>> Kafka source during a "backfill"- i.e. playing from an earlier Kafka
>> offset. While testing, we noticed some unexpected behaviour in the
>> watermark advancement which was resolved by removing `withIdleness` from
>> our watermark strategy.
>>
>>
>> val watermarkStrategy = WatermarkStrategy
>>   .forBoundedOutOfOrderness(Duration.ofMinutes(1))
>>   .withTimestampAssigner(new TimestampedEventTimestampAssigner[Event])
>>   .withWatermarkAlignment("alignment-group-1", Duration.ofMinutes(1))
>>   .withIdleness(Duration.ofMinutes(5))
>>
>> I have attached a couple of screenshots of the watermarkAlignmentDrift
>> metric. As you can see, the behaviour seems normal until a sudden drop in
>> the value to ~ Long.MIN_VALUE, which causes the pipeline to stop emitting
>> records completely from the source. Furthermore, the logs originating from
>> from
>> https://github.com/dawidwys/flink/blob/888162083fbb5f62f809e5270ad968db58cc9c5b/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinator.java#L174
>> also indicate that the `maxAllowedWatermark` switches to ~ Long.MIN_VALUE.
>>
>> We found that modifying the `updateInterval` passed into the alignment
>> parameters seemed to correlate with how long the pipeline would operate
>> before stopping - a larger interval of 20 minutes would encounter the issue
>> later than an interval of 1 second.
>>
>> We are wondering if a bug exists when using both `withIdleness` and
>> `withWatermarkAlignment`. Might it be related to
>> https://issues.apache.org/jira/browse/FLINK-28975, or is there possibly
>> a race condition in the watermark emission? We do not necessarily need to
>> have both configured at the same time, but we were also surprised by the
>> behaviour of the application. Has anyone run into a similar issue or have
>> further insight?
>>
>> Much Appreciated,
>> - Reem
>>
>>
>>
>>

Re: Jobs running in flink are not viable in YARN resource manager WEB UI

[Martijn Visser]

Hi,

I don't see anything in the 1.10 release notes [1] that would indicate that
anything has changed between 1.9 and 1.10 that would indicate that
something has changed on this topic. It should work.

Best regards,

Martijn

[1]
https://nightlies.apache.org/flink/flink-docs-release-1.10/release-notes/flink-1.10.html

On Tue, Mar 28, 2023 at 4:40 PM Kiran Kumar Kathe <
kirankumarkathe...@gmail.com> wrote:

> Actually we are using 1.10.1 in YARN cluster in production , so to make
> changes in cluster it is  bit complicated and it need to check all the jobs
> from first and need to  test everything , so  is there any chance of
> running the with the flink 1.10.1 by doing any configuration changes , that
> make jobs visible in yarn Web UI.
>
> On Tue, 28 Mar 2023 at 19:59, Martijn Visser 
> wrote:
>
>> Hi,
>>
>> You can't mix and match different versions; they all need to be the same
>> version. Flink 1.9.3 is no longer supported by the Flink community, I would
>> recommend upgrading to a still supported version (which is currently Flink
>> 1.16 and Flink 1.17)
>>
>> Best regards,
>>
>> Martijn
>>
>> On Tue, Mar 28, 2023 at 11:44 AM Kiran Kumar Kathe <
>> kirankumarkathe...@gmail.com> wrote:
>>
>>> When I submit jobs to flink using version 1.9.3 those jobs are
>>> visible in YARN resource manager Web UI , but when I use flink version
>>> 1.10.1  jobs running in flink are not visible in the Web UI , So in order
>>> to check where the issue is raising , I started replacing flink 1.10.1
>>> folders  with flink 1.9.3 then , I get to know that lib folder can one jar
>>> file named flink_dist which is responsible , to make  the jobs running in
>>> the flink visible in YARN resource manager. So my question here is Can I
>>> use the flink_dist jar of version 1.9.3 in place of  flink_dist 1.10.1 , if
>>> not can anyone please give me clarity what configuration changes should i
>>> do in order to make jobs running visible in YARN resource manager Web UI
>>> with Flink version 1.10.1. [ I observed that while I am using flink 1.9.3
>>> it is starting a yarn-session and I can see logs related to that and when
>>> using flink 1.10.1 there are no logs related to yarn-session are visible.]
>>>
>>

Re: [External] Re: Way to add columns with defaults to the existing table and recover from the savepoint

[Martijn Visser]

You could consider trying out the experimental version upgrade that was
introduced as part of FLIP-190: https://cwiki.apache.org/confluence/x/KZBnCw

On Tue, Mar 21, 2023 at 12:11 PM Ashish Khatkar via user <
user@flink.apache.org> wrote:

> Hi Shammon,
>
> Schema evolution works with avro type state, and Flink Table API uses
> RowData and has a serializer (RowDataSerializer) for it which doesn't
> allow change in column structure. Regarding state processor api, we are not
> creating any state in our service, we simply use Flink sql as a blackbox
> and let it handle the state. We simply create sql tables out of avro schema
> in StreamTableEnvironment and run queries on those sql tables by creating
> StatementSet and calling the execute() on it.
>
> I found flink doc on upgrade and evolution [1] and according to the doc it
> is not possible to achieve this.
>
> [1]
> https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/table/concepts/overview/#stateful-upgrades-and-evolution
>
> Cheers
> Ashish
>
> On Tue, Mar 21, 2023 at 1:51 AM Shammon FY  wrote:
>
>> Hi Ashish
>>
>> State compatibility is a complex issue, and you can review the state
>> evolution [1] and state processor [2] docs to see if there's a solution for
>> your problem.
>>
>> [1]
>> https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/fault-tolerance/serialization/schema_evolution/
>> [2]
>> https://nightlies.apache.org/flink/flink-docs-master/docs/libs/state_processor_api/
>>
>> Best,
>> Shammon FY
>>
>>
>> On Fri, Mar 17, 2023 at 8:48 PM Ashish Khatkar via user <
>> user@flink.apache.org> wrote:
>>
>>> Hi all,
>>>
>>> I need help in understanding if we can add columns with defaults, let's
>>> say NULL to the existing table and recover the job from the savepoint.
>>>
>>> We are using flink-1.16.0 table API and RocksDB as backend to provide a
>>> service to our users to run sql queries. The tables are created using the
>>> avro schema and when the schema is changed in a compatible manner i.e
>>> adding a field with default, we are unable to recover the job from the
>>> savepoint. This is the error we get after the schema is upgraded.
>>>
>>> Caused by: org.apache.flink.util.StateMigrationException: The new state 
>>> serializer 
>>> (org.apache.flink.table.runtime.typeutils.RowDataSerializer@aad5b03a) must 
>>> not be incompatible with the old state serializer 
>>> (org.apache.flink.table.runtime.typeutils.RowDataSerializer@9d089984).
>>>
>>> We tried to debug the issue and this error originates from
>>>
>>> org.apache.flink.table.runtime.typeutils.RowDataSerializer.RowDataSerializerSnapshot
>>>  -> resolveSchemaCompatibility line 343:345
>>>
>>> which checks the length of the type array and also the logicalType for
>>> each element or you can say columns.
>>>
>>> Is there a way to restore and evolve a table using table-api when the
>>> avro schema evolves in a compatible manner? If not, is there any plan to
>>> provide upgrades and evolutions with table apis?
>>>
>>> Cheers,
>>> Ashish Khatkar
>>>
>>

Re: Flink 1.17 upgrade issue when using azure storage account for checkpoints/savepoints

[Martijn Visser]

Hi,

This is tracked under https://issues.apache.org/jira/browse/FLINK-31612 and
a fix has been merged and will be made available when the first patch
version for Flink 1.16.1 will be released.

Best regards,

Martijn

On Sat, Mar 25, 2023 at 9:37 AM ChangZhuo Chen (陳昌倬) 
wrote:

> On Sat, Mar 25, 2023 at 02:01:24PM +0530, Jessy Ping wrote:
> > Root cause:  Caused by: java.util.concurrent.CompletionException:
> > java.lang.RuntimeException:  java.lang.ClassNotFoundException: Class
> > org.apache.hadoop.security.JniBasedUnixGroupsMappingWithFallback not
> found
>
> We have similar error with Google Cloud Storage, and there is workaround
> in slack thread
> https://apache-flink.slack.com/archives/C03G7LJTS2G/p1679320815257449
>
>
> --
> ChangZhuo Chen (陳昌倬) czchen@{czchen,debian}.org
> http://czchen.info/
> Key fingerprint = BA04 346D C2E1 FE63 C790  8793 CC65 B0CD EC27 5D5B
>