(DEPRECATED) Apache Flink User Mailing List archive.

RE: Flink kafka exceptions handling

Classic

List

Threaded

5 messages Options

BELGHITH Amira (EXT)

RE: Flink kafka exceptions handling

Thank you for your answer.

I have been subscribed.

This is the previous topic I’m referring to http://mail-archives.apache.org/mod_mbox/flink-user/202008.mbox/%3CCACzKVZQ093HixMewb_prtP41ceXgmxCv=CmPSBPHw-9+h8B4nA@...%3E

Our flink job manager fails after multiple restarting, when the Kafka Consumer does not find a topic for example. We have a kafka exception TopicUnthaurizationException. We listen to a list a topics and whenever one is down , all our streaming system is down .. is there a way to handle those exceptions in the FlinkKafkaConsumer so the job manager does not fail?

De : Amira Belghith <[hidden email]>
Envoyé : mercredi 6 janvier 2021 18:36
À : BELGHITH Amira (EXT) ResgGtsOpmOptVdf <[hidden email]>; [hidden email]
Objet : Fwd: Flink kafka exceptions handling

[EMETTEUR EXTERNE] / [EXTERNAL SENDER]
Soyez vigilant avant d'ouvrir les pièces jointes ou de cliquer sur les liens. En cas de doute, signalez le message via le bouton "Message suspect" ou consultez go/secu.
Be cautious before opening attachments or clicking on any links. If in doubt, use "Suspicious email" button or visit go/secu.

---------- Message transféré ---------
De : Piotr Nowojski <[hidden email]>
Date : mer. 6 janv. 2021 à 17:26
Objet : Re: Flink kafka exceptions handling
À : Amira Belghith <[hidden email]>
CC : [hidden email] <[hidden email]>

I think you first need to be subscribed as it's explained here [1]. Could you also link to which previous topic are you referring to?

Piotrek

[1] https://flink.apache.org/community.html#mailing-lists

śr., 6 sty 2021 o 17:09 Amira Belghith <[hidden email]> napisał(a):

Hey,

Thanks for your fast reply.

The mail couldnt be delivered to the mailing list.

Le mer. 6 janv. 2021 à 16:59, Piotr Nowojski <[hidden email]> a écrit :

Hey,

could you post the question on the user <[hidden email]> mailing list?

Thanks,

Piotrek

śr., 6 sty 2021 o 15:11 Amira Belghith <[hidden email]> napisał(a):

Hi Nick, Piotr,

Im a software engineer working for Societe Generale bank.

I saw your discussion about FlinkKafkaConsumer and exceptions handling.

I have the same problem for a week now, and I wanted to know if you have found a solution.

Our flink job manager fails after multiple restarting, when the Kafka Consumer does not find a topic for example. We have a kafka exception TopicUnthaurizationException. We listen to a list a topics and whenever one is down , all our streaming system is down .. is there a way to handle those exceptions in the FlinkKafkaConsumer so the job manager does not fail?

Thanks a lot for your help,

Amira belghith

=========================================================

Ce message et toutes les pieces jointes (ci-apres le "message")
sont confidentiels et susceptibles de contenir des informations
couvertes par le secret professionnel. Ce message est etabli
a l'intention exclusive de ses destinataires. Toute utilisation
ou diffusion non autorisee interdite.
Tout message electronique est susceptible d'alteration. La SOCIETE GENERALE
et ses filiales declinent toute responsabilite au titre de ce message
s'il a ete altere, deforme falsifie.

=========================================================

This message and any attachments (the "message") are confidential,
intended solely for the addresses, and may contain legally privileged
information. Any unauthorized use or dissemination is prohibited.
E-mails are susceptible to alteration. Neither SOCIETE GENERALE nor any
of its subsidiaries or affiliates shall be liable for the message
if altered, changed or falsified.

=========================================================

Piotr Nowojski-4

Re: Flink kafka exceptions handling

Hi Amira,

I think the previous topic you are referring to doesn't seem to be related with your current problem.

Regarding your problem, I'm afraid I don't know the FlinkKafkaConsumer code too well. Maybe someone else from the community could help?

Best,

Piotrek

śr., 6 sty 2021 o 19:01 BELGHITH Amira (EXT) <[hidden email]> napisał(a):

Thank you for your answer.

I have been subscribed.

This is the previous topic I’m referring to http://mail-archives.apache.org/mod_mbox/flink-user/202008.mbox/%3CCACzKVZQ093HixMewb_prtP41ceXgmxCv=CmPSBPHw-9+h8B4nA@...%3E

Our flink job manager fails after multiple restarting, when the Kafka Consumer does not find a topic for example. We have a kafka exception TopicUnthaurizationException. We listen to a list a topics and whenever one is down , all our streaming system is down .. is there a way to handle those exceptions in the FlinkKafkaConsumer so the job manager does not fail?

De : Amira Belghith <[hidden email]>
Envoyé : mercredi 6 janvier 2021 18:36
À : BELGHITH Amira (EXT) ResgGtsOpmOptVdf <[hidden email]>; [hidden email]
Objet : Fwd: Flink kafka exceptions handling

[EMETTEUR EXTERNE] / [EXTERNAL SENDER]
Soyez vigilant avant d'ouvrir les pièces jointes ou de cliquer sur les liens. En cas de doute, signalez le message via le bouton "Message suspect" ou consultez go/secu.
Be cautious before opening attachments or clicking on any links. If in doubt, use "Suspicious email" button or visit go/secu.

---------- Message transféré ---------
De : Piotr Nowojski <[hidden email]>
Date : mer. 6 janv. 2021 à 17:26
Objet : Re: Flink kafka exceptions handling
À : Amira Belghith <[hidden email]>
CC : [hidden email] <[hidden email]>

I think you first need to be subscribed as it's explained here [1]. Could you also link to which previous topic are you referring to?

Piotrek

[1] https://flink.apache.org/community.html#mailing-lists

śr., 6 sty 2021 o 17:09 Amira Belghith <[hidden email]> napisał(a):

Hey,

Thanks for your fast reply.

The mail couldnt be delivered to the mailing list.

Le mer. 6 janv. 2021 à 16:59, Piotr Nowojski <[hidden email]> a écrit :

Hey,

could you post the question on the user <[hidden email]> mailing list?

Thanks,

Piotrek

śr., 6 sty 2021 o 15:11 Amira Belghith <[hidden email]> napisał(a):

Hi Nick, Piotr,

Im a software engineer working for Societe Generale bank.

I saw your discussion about FlinkKafkaConsumer and exceptions handling.

I have the same problem for a week now, and I wanted to know if you have found a solution.

Our flink job manager fails after multiple restarting, when the Kafka Consumer does not find a topic for example. We have a kafka exception TopicUnthaurizationException. We listen to a list a topics and whenever one is down , all our streaming system is down .. is there a way to handle those exceptions in the FlinkKafkaConsumer so the job manager does not fail?

Thanks a lot for your help,

Amira belghith

=========================================================

Ce message et toutes les pieces jointes (ci-apres le "message")
sont confidentiels et susceptibles de contenir des informations
couvertes par le secret professionnel. Ce message est etabli
a l'intention exclusive de ses destinataires. Toute utilisation
ou diffusion non autorisee interdite.
Tout message electronique est susceptible d'alteration. La SOCIETE GENERALE
et ses filiales declinent toute responsabilite au titre de ce message
s'il a ete altere, deforme falsifie.

=========================================================

This message and any attachments (the "message") are confidential,
intended solely for the addresses, and may contain legally privileged
information. Any unauthorized use or dissemination is prohibited.
E-mails are susceptible to alteration. Neither SOCIETE GENERALE nor any
of its subsidiaries or affiliates shall be liable for the message
if altered, changed or falsified.

=========================================================

Aljoscha Krettek

Re: Flink kafka exceptions handling

In reply to this post by BELGHITH Amira (EXT)

Hi,

When you say that the `JobManager` goes down, you're referring to the
fact that the Flink job will finish in a failed state after too many
exceptions have occurred in the `FlinkKafkaConsumer. Is that correct?

I'm afraid right now there is no code path that would allow catching
those `TopicUnthaurizationException`. We basically treat most exceptions
coming from Kafka as errors that require recovery.

What behaviour would you have in mind as a reaction to those exceptions?

Best,
Aljoscha

On 2021/01/06 17:59, BELGHITH Amira (EXT) wrote:

>
>Thank you for your answer.
>I have been subscribed.
>
>This is the previous topic I’m referring to http://mail-archives.apache.org/mod_mbox/flink-user/202008.mbox/%3CCACzKVZQ093HixMewb_prtP41ceXgmxCv=CmPSBPHw-9+h8B4nA@...%3E
>
>Our flink job manager fails after multiple restarting, when the Kafka Consumer does not find a topic for example. We have a kafka exception TopicUnthaurizationException. We listen to a list a topics and whenever one is down , all our streaming system is down .. is there a way to handle those exceptions in the FlinkKafkaConsumer so the job manager does not fail?
>
>
>De : Amira Belghith <[hidden email]>
>Envoyé : mercredi 6 janvier 2021 18:36
>À : BELGHITH Amira (EXT) ResgGtsOpmOptVdf <[hidden email]>; [hidden email]
>Objet : Fwd: Flink kafka exceptions handling
>
>[EMETTEUR EXTERNE] / [EXTERNAL SENDER]
>Soyez vigilant avant d'ouvrir les pièces jointes ou de cliquer sur les liens. En cas de doute, signalez le message via le bouton "Message suspect" ou consultez go/secu.
>Be cautious before opening attachments or clicking on any links. If in doubt, use "Suspicious email" button or visit go/secu.
>
>
>
>---------- Message transféré ---------
>De : Piotr Nowojski <[hidden email]<mailto:[hidden email]>>
>Date : mer. 6 janv. 2021 à 17:26
>Objet : Re: Flink kafka exceptions handling
>À : Amira Belghith <[hidden email]<mailto:[hidden email]>>
>CC : [hidden email]<mailto:[hidden email]> <[hidden email]<mailto:[hidden email]>>
>
>I think you first need to be subscribed as it's explained here [1]. Could you also link to which previous topic are you referring to?
>
>Piotrek
>
>[1] https://flink.apache.org/community.html#mailing-lists
>
>śr., 6 sty 2021 o 17:09 Amira Belghith <[hidden email]<mailto:[hidden email]>> napisał(a):
>Hey,
>Thanks for your fast reply.
>The mail couldnt be delivered to the mailing list.
>
>Le mer. 6 janv. 2021 à 16:59, Piotr Nowojski <[hidden email]<mailto:[hidden email]>> a écrit :
>Hey,
>
>could you post the question on the user <[hidden email]<mailto:[hidden email]>> mailing list?
>
>Thanks,
>Piotrek
>
>śr., 6 sty 2021 o 15:11 Amira Belghith <[hidden email]<mailto:[hidden email]>> napisał(a):
>Hi Nick, Piotr,
>
>Im a software engineer working for Societe Generale bank.
>I saw your discussion about FlinkKafkaConsumer and exceptions handling.
>I have the same problem for a week now, and I wanted to know if you have found a solution.
>Our flink job manager fails after multiple restarting, when the Kafka Consumer does not find a topic for example. We have a kafka exception TopicUnthaurizationException. We listen to a list a topics and whenever one is down , all our streaming system is down .. is there a way to handle those exceptions in the FlinkKafkaConsumer so the job manager does not fail?
>
>Thanks a lot for your help,
>Amira belghith
>
>=========================================================
>
>Ce message et toutes les pieces jointes (ci-apres le "message")
>sont confidentiels et susceptibles de contenir des informations
>couvertes par le secret professionnel. Ce message est etabli
>a l'intention exclusive de ses destinataires. Toute utilisation
>ou diffusion non autorisee interdite.
>Tout message electronique est susceptible d'alteration. La SOCIETE GENERALE
>et ses filiales declinent toute responsabilite au titre de ce message
>s'il a ete altere, deforme falsifie.
>
>=========================================================
>
>This message and any attachments (the "message") are confidential,
>intended solely for the addresses, and may contain legally privileged
>information. Any unauthorized use or dissemination is prohibited.
>E-mails are susceptible to alteration. Neither SOCIETE GENERALE nor any
>of its subsidiaries or affiliates shall be liable for the message
>if altered, changed or falsified.
>
>=========================================================

BELGHITH Amira (EXT)

RE: Flink kafka exceptions handling

In reply to this post by BELGHITH Amira (EXT)

Thanks for your feedbacks.

Please find below my answers:

-----Message d'origine-----

De : Aljoscha Krettek <[hidden email]>

Envoyé : jeudi 7 janvier 2021 13:55

À : [hidden email]

Objet : Re: Flink kafka exceptions handling

[EMETTEUR EXTERNE] / [EXTERNAL SENDER]

Soyez vigilant avant d'ouvrir les pièces jointes ou de cliquer sur les liens. En cas de doute, signalez le message via le bouton "Message suspect" ou consultez go/secu.

Be cautious before opening attachments or clicking on any links. If in doubt, use "Suspicious email" button or visit go/secu.

Hi,

When you say that the `JobManager` goes down, you're referring to the fact that the Flink job will finish in a failed state after too many exceptions have occurred in the `FlinkKafkaConsumer. Is that correct?

--> Yes, this is exactly what I meant, the Flink job is in a failed state

I'm afraid right now there is no code path that would allow catching those `TopicUnthaurizationException`. We basically treat most exceptions coming from Kafka as errors that require recovery.

--> We can have other exceptions, like like TimeoutException when our Kafka brokers are degraded.

What behaviour would you have in mind as a reaction to those exceptions?

--> Our processing System is supposed to continue streaming data even though there is some Kafka errors, we are expecting that the KafkaConsumer fails but not the Flink job, do you think it is possible?

Best,

Aljoscha

On 2021/01/06 17:59, BELGHITH Amira (EXT) wrote:

>Thank you for your answer.

>I have been subscribed.

>This is the previous topic I’m referring to

>http://mail-archives.apache.org/mod_mbox/flink-user/202008.mbox/%3CCACz

>KVZQ093HixMewb_prtP41ceXgmxCv=[hidden email]%3E

>Our flink job manager fails after multiple restarting, when the Kafka Consumer does not find a topic for example. We have a kafka exception TopicUnthaurizationException. We listen to a list a topics and whenever one is down , all our streaming system is down .. is there a way to handle those exceptions in the FlinkKafkaConsumer so the job manager does not fail?

>De : Amira Belghith <[hidden email]> Envoyé : mercredi 6

>janvier 2021 18:36 À : BELGHITH Amira (EXT) ResgGtsOpmOptVdf

><[hidden email]>; [hidden email] Objet :

>Fwd: Flink kafka exceptions handling

>[EMETTEUR EXTERNE] / [EXTERNAL SENDER]

>Soyez vigilant avant d'ouvrir les pièces jointes ou de cliquer sur les liens. En cas de doute, signalez le message via le bouton "Message suspect" ou consultez go/secu.

>Be cautious before opening attachments or clicking on any links. If in doubt, use "Suspicious email" button or visit go/secu.

>---------- Message transféré ---------

>De : Piotr Nowojski <[hidden email]<mailto:[hidden email]>>

>Date : mer. 6 janv. 2021 à 17:26

>Objet : Re: Flink kafka exceptions handling À : Amira Belghith

><[hidden email]<mailto:[hidden email]>>

>CC : [hidden email]<mailto:[hidden email]>

><[hidden email]<mailto:[hidden email]>>

>I think you first need to be subscribed as it's explained here [1]. Could you also link to which previous topic are you referring to?

>Piotrek

>[1] https://flink.apache.org/community.html#mailing-lists

>śr., 6 sty 2021 o 17:09 Amira Belghith <[hidden email]<mailto:[hidden email]>> napisał(a):

>Hey,

>Thanks for your fast reply.

>The mail couldnt be delivered to the mailing list.

>Le mer. 6 janv. 2021 à 16:59, Piotr Nowojski <[hidden email]<mailto:[hidden email]>> a écrit :

>Hey,

>could you post the question on the user <[hidden email]<mailto:[hidden email]>> mailing list?

>Thanks,

>Piotrek

>śr., 6 sty 2021 o 15:11 Amira Belghith <[hidden email]<mailto:[hidden email]>> napisał(a):

>Hi Nick, Piotr,

>Im a software engineer working for Societe Generale bank.

>I saw your discussion about FlinkKafkaConsumer and exceptions handling.

>I have the same problem for a week now, and I wanted to know if you have found a solution.

>Thanks a lot for your help,

>Amira belghith

>=========================================================

>Ce message et toutes les pieces jointes (ci-apres le "message") sont

>confidentiels et susceptibles de contenir des informations couvertes

>par le secret professionnel. Ce message est etabli a l'intention

>exclusive de ses destinataires. Toute utilisation ou diffusion non

>autorisee interdite.

>Tout message electronique est susceptible d'alteration. La SOCIETE

>GENERALE et ses filiales declinent toute responsabilite au titre de ce

>message s'il a ete altere, deforme falsifie.

>=========================================================

>This message and any attachments (the "message") are confidential,

>intended solely for the addresses, and may contain legally privileged

>information. Any unauthorized use or dissemination is prohibited.

>E-mails are susceptible to alteration. Neither SOCIETE GENERALE nor any

>of its subsidiaries or affiliates shall be liable for the message if

>altered, changed or falsified.

>=========================================================

=========================================================

Aljoscha Krettek

Re: Flink kafka exceptions handling

On 2021/01/07 14:36, BELGHITH Amira (EXT) wrote:
>--> Our processing System is supposed to continue streaming data even
>though there is some Kafka errors, we are expecting that the
>KafkaConsumer fails but not the Flink job, do you think it is possible?

I'm afraid that's not possible with Flink right no. We treat all
exceptions as errors, which lead to job restarts and eventually complete
job failure if the restarts exceed the configured limit.

What you could do right now is copy the code for the
`FlinkKafkaConsumer` and insert exception handling code for the
exceptions that you would like to exclude. You could even go so far and
add generic handling code that you can then configure with a list of
exceptions to ignore when creating the consumer.

I hope that helps!

Best,
Aljoscha