Re: New blog post published - Sort-Based Blocking Shuffle Implementation in Flink

Thank Daisy& Kevin much for your introduction to the improvement of TM blocking shuffle, credit base+io scheduling is indeed a very interesting thing. At the same time, I look forward to this as a default setting for tm blocking shuffle. Best, Guowei On Wed, Nov 3, 2021 at 2:46 PM Gen Luo wrote

Re: What is Could not retrieve file from transient blob store?

Hi, Smith It seems that the log file(blob_t-274d3c2d5acd78ced877d898b1877b10b62a64df- 590b54325d599a6782a77413691e0a7b) is deleted for some reason. But AFAIK there are no other guys reporting this exception.(Maybe other guys know what would happen). 1. I think if you could refresh the page and you

Re: New blog post published - Sort-Based Blocking Shuffle Implementation in Flink

Thanks Daisy and Kevin! The benchmark results look really exciting! On Tue, Nov 2, 2021 at 4:38 PM David Morávek wrote: > Thanks Daisy and Kevin for a great write up! ;) Especially the 2nd part > was really interesting, I really like the idea of the single spill file > with a custom scheduling o

Re: Data Stream countWindow followed by keyBy does not preserve time order

Hi I did not run your program directly, but I see that you are now using the Batch execution mode. I suspect it is related to this, because in the Batch execution mode FLINK will "sort" the Key (this might be an unstable sort). So would you like to experiment with the results of running with Strea

Re: Flink sink data to DB and then commit data to Kafka

Hi, Qihua AFAIK there is no way to do it. Maybe you need to implement a "new" sink to archive this target. Best, Guowei On Wed, Nov 3, 2021 at 12:40 PM Qihua Yang wrote: > Hi, > > Our flink application has two sinks(DB and kafka topic). We want to push > same data to both sinks. Is it possibl

Flink sink data to DB and then commit data to Kafka

Hi, Our flink application has two sinks(DB and kafka topic). We want to push same data to both sinks. Is it possible to push data to kafka topic only after data is pushed to DB successfully? If the commit to DB fail, we don't want those data is pushed to kafka. Thanks, Qihua

Why do the count windows in Flink Table APIs require processing time for sorting whereas in Flink Datastream APIs they do not

I have read about the Window operator in Flink documentation and know that it groups rows into finite groups based on time or row-count intervals. I saw an example of a sliding count window right

Re: Flink + K8s

Hi Rommel, That’s correct that K8s will restart the JM pod (assuming it’s been created by a K8s Job or Deployment), and it will pick up the HA data and resume work. The only use case for having multiple replicas is faster failover, so you don’t have to wait for K8s to provision that new pod (which

What is Could not retrieve file from transient blob store?

Hi running Flink 1.10.0 With 3 zookeepers, 3 job nodes and 3 task nodes. and I saw this exception on the job node logs... 2021-11-02 23:20:22,703 ERROR org.apache.flink.runtime.rest.handler.taskmanager.TaskManagerLogFileHandler - Unhandled exception. org.apache.flink.util.FlinkException: Could not

Re: Question on BoundedOutOfOrderness

Hi Oliver, I believe you also need to do sort, out of order ness watermark strategy only “postpone” watermark for given expected maximum of out of orderness. Check Ververica example - https://github.com/ververica/flink-training-exercises/blob/master/src/main/java/com/ververica/flinktraining/exam

Re: How to refresh topics to ingest with KafkaSource?

Hi Arvid, I have some bandwidth to contribute to this task and am familiar with the code. Could you or another committer assign me this ticket? Thanks, Mason > On Oct 30, 2021, at 5:24 AM, Arvid Heise wrote: > > Hi Mason, > > thanks for creating that. > > We are happy to take contribuitons

Re: Possibility of supporting Reactive mode for native Kubernetes application mode

I tried to remove the Re: [EXTERNAL] label just now, but it seems to break the thread on my side. I guess I can’t do much on my side regarding this. The email is forced to add such tag at the company level. Anyways, l guess we can continue to discuss the issue in this thread. Thanks, Fuyao Fro

Is there a way to update checkpoint configuration for a job "in-place"?

Hi all, We run a Flink application on Kubernetes in Application Mode using Kafka with exactly-once-semantics and high availability. We are looking into a specific failure scenario: a flink job that has too short a checkpoint timeout (execution.checkpointing.timeout) and at some point during the j

Re: Possibility of supporting Reactive mode for native Kubernetes application mode

Hi David, Nicolaus, Thanks for the reply. 1. For your first question, Yes. I want to use the checkpoint to stop and restart the application. I think this is similar to the Reactive mode strategy, right? (I don’t know the exact implementation behind the Reactive mode). From your description

Flink + K8s

Hi, >From my understanding, when i set Flink in HA mode in K8s, I don't need to setup more than 1 job manager, because once the job manager dies, K8s will restart it for me. Is that the correct understanding or for the HA purpose, I still need to setup more than 1 job manager? Thanks. Rommel --

Question on BoundedOutOfOrderness

Hi! I am investigating the use of Flink for a new project and started some simple demos. Currently I am stuck at the point where I need to deal with events arriving out of order based on their event time. I’ve spent quite some time researching on SO, the docs, the Ververica training (excellen

Re: NoClassDefFoundError for RMIServerImpl when running Flink with Scala 2.12 and Java 11

Hi Nicolaus, Thanks for your reply. It turns out to be the Java distribution used in the base image. I changed the base image and it works now. On Tue, Nov 2, 2021 at 10:14 AM Nicolaus Weidner wrote: > > Hi, > > I tried building Flink 1.13 with the Scala 2.12 profile and running some of > the

Re: NoClassDefFoundError for RMIServerImpl when running Flink with Scala 2.12 and Java 11

Hi, I tried building Flink 1.13 with the Scala 2.12 profile and running some of the examples with Java 11, without encountering the issue you describe (with or without HA). Can you give more details on how exactly you built Flink (ideally the full command), and how you ran the job? Best, Nico On

Data Stream countWindow followed by keyBy does not preserve time order

Hi all, Can anyone advise on this? I wrote a simple test of the countWindow method (in Kotlin) as below package aero.airlab.flinkjobs.headingreminder import org.apache.flink.api.common.RuntimeExecutionMode import org.apache.flink.api.common.eventtime.WatermarkStrategy import org.apache.flink.st

Re: Reactive mode in 1.13

Hi Ravi, I think you also need to make the tornado.jar available to the TaskExecutor processes (e.g. putting them into the usrlib or lib directory where you started the process). When using the application mode, then Flink assumes that all processes have access to the user code jar. That's why Fli

Re: Possibility of supporting Reactive mode for native Kubernetes application mode

Hi Fuyao, About your second question: You are right that taking and restoring from savepoints will incur a performance loss. They cannot be incremental, and cannot use native (low-level) data formats - for now. These issues are on the list of things to improve for Flink 1.15, so if the changes mak

Statefun remote functions - acessing kafka headers from a remote function

Hi, is there a neat way to access kafka headers from within a remote function without using the datastream api to insert the headers as part of a RoutableMessage payload? Many thanks Fil

Re: Statefun embedded functions - parallel per partition, sequential per key

Hi All Just an update for future reference, it turned out that the machine we were using for this test didn't have enough memory for what we were asking it to do. It was that simple. The upside is that not even with the world's most unstable cluster did we manage to lose a single message. Just as

Re: [External] : Re: Possibility of supporting Reactive mode for native Kubernetes application mode

> > Similar to Reactive mode, checkpoint must be enabled to support such > functionality. ... > Wouldn't that mean tearing down the whole Flink cluster in order to re-scale? That could be quite costly. We're aiming to speed-up the recovery process for the reactive mode and this would most likely b

Re: Upgrading to the new KafkaSink + Flink 1.14 from Flink 1.13.2 + FlinkKafkaProducer

Hi Yuval, Ok no worries. One thing I would first check is why the TwoPhaseCommitSinkFunction is instantiated because the KafkaSink is not using it. It seems there is still an old FlinkKafkaProducer build somewhere. Best, Fabian

Re: Upgrading to the new KafkaSink + Flink 1.14 from Flink 1.13.2 + FlinkKafkaProducer

Hi Fabian, The program is part of a very large Flink infrastructure we have in place so unfortunately sharing I can't share it. But perhaps point me to more concrete information you'd like to receive? On Tue, Nov 2, 2021 at 1:51 PM Fabian Paul wrote: > Hi Yuval, > > This error looks indeed stra

Re: Upgrading to the new KafkaSink + Flink 1.14 from Flink 1.13.2 + FlinkKafkaProducer

Hi Yuval, This error looks indeed strange. I do not think when switching to the unified KafkaSink the old serializer should be invoked at all. Can you maybe share more information about the job you are using or maybe share the program so that we can reproduce it? Best, Fabian

Re: Converting a Table to a DataStream[RowData] instead of DataStream[Row] with `toDataStream`

It works! Thank you Nicolaus. On Tue, Nov 2, 2021 at 12:40 PM Nicolaus Weidner < nicolaus.weid...@ververica.com> wrote: > Hi Yuval, > > Can you try > toDataStream[RowData](tableSchema.toPhysicalRowDataType.bridgedTo(classOf[RowData]))? > > Best regards, > Nico > > On Thu, Oct 28, 2021 at 10:15 P

Upgrading to the new KafkaSink + Flink 1.14 from Flink 1.13.2 + FlinkKafkaProducer

Hi, I'm trying to upgrade an application from Flink 1.14 + using the new KafkaSink while restoring from a checkpoint. I changed the UID of the KafkaSink completely and ran the application with --allow-non-restored-state. However, when restoring, I still run into the following error: Caused by: or

Re: Converting a Table to a DataStream[RowData] instead of DataStream[Row] with `toDataStream`

Hi Yuval, Can you try toDataStream[RowData](tableSchema.toPhysicalRowDataType.bridgedTo(classOf[RowData]))? Best regards, Nico On Thu, Oct 28, 2021 at 10:15 PM Yuval Itzchakov wrote: > Flink 1.14 > Scala 2.12.5 > > Hi, > I want to be able to convert a Table into a DataStream[RowData]. I need t

Re: Reactive mode in 1.13

Hi Ravi, The reactive mode shouldn't do things differently compared to a normal application cluster deployment. Maybe you can show us exactly how you submit a job, the contents of the bundled jar, how you build the fat jar and the logs of the failed Flink run. Moving this discussion to the user M

Re: New blog post published - Sort-Based Blocking Shuffle Implementation in Flink

Thanks Daisy and Kevin for a great write up! ;) Especially the 2nd part was really interesting, I really like the idea of the single spill file with a custom scheduling of read requests. Best, D. On Mon, Nov 1, 2021 at 10:01 AM Daisy Tsang wrote: > Hey everyone, we have a new two-part post publ

Re: Kryo Serialization issues in Flink Jobs.

Hi Prasanna, it could be a bug where the ExecutionConfig is not forwarded properly to all locations where the KryoSerializer is used. As a first step for debugging, I would recommend to create a custom TypeInformation (most methods are not relevant except for createTypeSerializer and getType