Re: How to gracefully handle job recovery failures

Hi Li, If I understand correctly, you want the cluster to proceed recovery, skipping some non-recoverable jobs (but still recover others). The only way I can think of is to remove the corresponding nodes in ZooKeeper which is not very safe. I'm pulling in Robert and Till who might know better. R

Re: confused about `TO_TIMESTAMP` document description

Hi Leonard, Thanks for confirmation. I have created the jira ticket [1]. The pull request will be submitted later. best regards, [1] https://issues.apache.org/jira/browse/FLINK-22970 Leonard Xu 於 2021年6月10日 週四 下午8:58寫道： > Hi，Tony > > > I found this code snippet [2] might be related to `TO_TIM

Re: Add control mode for flink

Thanks for all the discussions and suggestions. Since the topic has been discussed for about a week, it is time to have a conclusion and new ideas are welcomed at the same time. First, the topic starts with use cases in restful interface. The restful interface supported many useful interact

Re: Looking for online live training courses

Hi XiaQu, Welcome to Flink community! I don't know if there are online interactive training courses yet, I would add it to the email later after I consult offline. I would like to list the most popular resources I know as follows, I hope it helps. Training Course 1. https://flink.apache.org/traini

Looking for online live training courses

Hi all, My team is planning to start our journey of Apache Flink, was wondering if there are any professional training courses (online interactive at this moment) recommended? Thanks Best, *Xia(Nate) Qu*

[table-walkthrough] docker-compose build issues

Hi, *I have successfully run the project "table-walkthrough" on IDEA (w/t errors but warnings)*, *I'm now trying to build this project by using the "docker-compose" command* as the tutorial does. However, when I run the "docker-compose build" command in the directory of project "table-walkthrough"

Kafka Connector Topic Discovery

I'm trying to use the topic discovery feature of the Kafka Connector. The problem I'm having is that Kafka Consumers fail to start if there are no topics matching the topic regex when they start up. Is this intended behavior? Is there some other property I could set to just continue discovery un

Flink and Avro for state serialization

Hello all, We are running some Flink jobs - we wrote the job in Beam but are using the Flink Runner and are seeing the following error when restarting the job from a Savepoint Caused by: java.io.InvalidClassException: com.xxx.xxx; local class > incompatible: stream classdesc serialVersionUID = -5

Re: How to gracefully handle job recovery failures

Hi Roman, Is there a way to abandon job recovery after a few tries? By that I mean that this problem was fixed by me restarting the cluster and not try to recover a job. Is there some setting that emulates what I did, so I don't need to do manual intervention if this happens again?? Thanks, Li O

Re: PyFlink: Upload resource files to Flink cluster

Hi, I think the second option is what you need. The documentation says only zip format is supported. Alternatively, you could upload the files to S3 or other DFS and access from TMs and re-upload when needed. [1] https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/dev/python/depende

Re: NPE when restoring from savepoint in Flink 1.13.1 application

Hi ChangZhuo, Thanks for reporting, it looks like a bug. I've opened a ticket for that [1]. [1] https://issues.apache.org/jira/browse/FLINK-22966 Regards, Roman On Wed, Jun 9, 2021 at 4:07 PM ChangZhuo Chen (陳昌倬) wrote: > > Hi, > > We have NullPointerException when trying to restore from savep

Re: How to gracefully handle job recovery failures

Hi Li, The missing file is a serialized job graph and the job recovery can't proceed without it. Unfortunately, the cluster can't proceed if one of the jobs can't recover. Regards, Roman On Thu, Jun 10, 2021 at 6:02 AM Li Peng wrote: > > Hey folks, we have a cluster with HA mode enabled, and re

Re: FileSource may cause akka.pattern.AskTimeoutException, and akka.ask.timeout not workable

Hi, I think you need to increase client.timeout [1]. Regarding the FileSource, it's difficult to say whether it is the reason. The logs you provided are from the client, JobManager logs would be helpful. [1] https://ci.apache.org/projects/flink/flink-docs-master/docs/deployment/config/#client-tim

Re: How to know (in code) how many times the job restarted?

Hi Felipe, You can use getRuntimeContext().getAttemptNumber() [1] (but beware that depending on the configuration only a pipeline region can be restarted, not the whole job). But if all you want is to check whether it's a first attempt or not, you can also call context.isRestored() from initializ

Re: Is the description of taskmanager.memory.task.heap.size in the official document incorrect?

Hi Jason, I think you are right, taskmanager.memory.framework.off-heap.size is also subtracted, at least according to the source code [1]. Would you like to create a jira issue? [1] https://github.com/apache/flink/blob/4cfdc314e39974eaf089ffe0512893b01643ed2e/flink-runtime/src/main/java/org/apach

How to know (in code) how many times the job restarted?

Hello community, Is it possible to know programmatically how many times my Flink stream job restarted since it was running? My use case is like this. I have an Unit test that uses checkpoint and I throw one exception in a MapFunction for a given time, i.e.: for the 2 seconds ahead. Because Flink

Re: Stateful Function Ingress issues

Hello Jessy, Currently StateFun Kafka ingress interprets the keys of the record as the destination address. So you'd have to attach a key to use that specific ingress. If this is not an option for you, you can consider @Tim's suggestion or create a JIRA with a feature request, which we will be ha

Re: Stateful Function Ingress issues

Hi Jessy, I had this issue as well, here's the resolution . I ended up forking the version of statefun I used and removing the null check to default to empty string, but I'm goin

Re: Stateful Function Ingress issues

Hi all, I am trying to consume data from azure eventhub using the kafka ingress and i am getting the following error. *java.lang.IllegalStateException: The io.statefun.kafka/ingress ingress requires a UTF-8 key set for each record.* While sending the data to the Event hub using my data producer,

Stateful Function Ingress issues

Hi all, I am trying to consume data from azure eventhub using the kafka ingress and i am getting the following error. *java.lang.IllegalStateException: The io.statefun.kafka/ingress ingress requires a UTF-8 key set for each record.* While sending the data to the Event hub using my data producer,

Re: confused about `TO_TIMESTAMP` document description

Hi，Tony > I found this code snippet [2] might be related to `TO_TIMESTAMP` udf, and > seems like it won't set use any timezone configuration, so maybe the document > might be wrong. Your analysis is right，the document is wrong, we should correct it. Would you like to create an jira ticket and

Is the description of taskmanager.memory.task.heap.size in the official document incorrect?

Hello everyone When I studied the memory model of TaskManager, I found that there is a problem in the official document, which is the description of taskmanager.memory.task.heap.size is incorrect. (https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/deployment/config/#taskmanag

Re: Persisting state in RocksDB

Hi Paul, You can leave operators dangling. So no need to add fake sinks. If you write to HTTP, the best option is actually asyncIO. [1] This will run much much faster. AsyncIO however has no state access (we want to change that eventually but for now it's to avoid too many antipatterns). For me

Re: question about concating an array in flink sql

Hi vtygoss, If the length of names is fixed, please try this 'select id, concat_ws(',',names[1],names[2],names[3]) from test', and note begin with 1 instead of 0. Else maybe you need to define a custom UDF which receives two arguments, first is string as separator, second is a string array as conte

Re: Persisting state in RocksDB

Hi Arvid - thanks for the welcome :) The SinkFunction is custom (extends RichSinkFunction), and writes to a legacy web service over HTTP. I’ll investigate the keyBy+KeyedProcessFunction further - thanks. Frankly I looked at this but I think I was confusing myself between working with KV store

question about concating an array in flink sql

Hi, I have below use case I want concat an array using comma separator, but got exception “Cannot apply 'CONCAT_WS' to arguments of type 'CONCAT_WS(, )'. Supported form(s): 'CONCAT_WS()’”。 How to concat an array in flink sql? please help to offer some advices. Regards ``` [test case