Even key distribution workload

Hi All, Currently I have a keyBy user and I see uneven load distribution since some of the users would have very high load versus some users having very few messages. Is there a recommended way to achieve even distribution of workload? Has someone else encountered this problem and what was the wor

Re: How to create Row with RowTypeInfo

I forget to add the user mailing list in the response. I now add user mailing list to the response in case other users might want to solve this problem too... Soheil Pourbafrani 于2019年7月15日周一 上午2:56写道： > Great! > I got it > > Thanks > > On Sun, Jul 14, 2019 at 8:26 PM Caizhi Weng wrote: > >> Hi

Re: Flink SQL API: Extra columns added from order by

(Oops, I mistakenly sent my response only to Rong Rong. Sorry Rong...) Hi Morrisa, This is due to a bug in the old flink planner. The `createTable(new PlannerQueryOperation(relational.rel))` method in flink-table-planner -> `TableEnvImpl` -> `sqlQuery` should be `createTable(new PlannerQueryOpera

Re: Graceful Task Manager Termination and Replacement

Hi Aaron, >From my understanding, you want shutting down a Task Manager without restart the job which has tasks running on this Task Manager? Based on current implementation, if there is a Task Manager is down, the tasks on it would be treated as failed. The behavior of task failure is defined vi

Re: Running Flink cluster via Marathon

How many Job manager did you configure ? On Mon, Jul 15, 2019, 07:11 Marzieh Ghasemy wrote: > Hello I have a Mesos cluster of two master and three slaves, I configured > Marathon and Zookeeper. My Zookeeper cluster has five nodes. When I run > Flink Json file via Marathon, it is run, but I ca

Running Flink cluster via Marathon

HelloI have a Mesos cluster of two master and three slaves, I configured Marathon and Zookeeper. My Zookeeper cluster has five nodes. When I run Flink Json file via Marathon, it is run, but I can see Flink UI in just one slave. Other slaves show me this error:Service temporarily unavailable due

Re: Checkpoint failure

Hi, It seems that sometimes your job can not finish checkpoint during 1 minute (timeout in your configuration). The checkpoint of 9867 is expired. The 9868 costs 50 seconds, almost reaches the timeout limit. However the 9869 and 9870 finished quite soon. It's more like a performance issue. So it'

Re: Apache Flink - Event time and process time timers with same timestamp

Also, are the event time timers and processing time timers handled separately - ie,  if I register event time timer and then use the same timestamp to delete processing time timer - will it remove the event time timer registration ?   In the example  https://github.com/streaming-with-flink/examp

Apache Flink - Event time and process time timers with same timestamp

Hi: If I register both event time timer and processing time timer with the same timestamp for a particular key - will they both fire or only one will fire ? If only one, what will be its time domain ? Thanks

[ANNOUNCE] Weekly Community Update 2019/28

Dear community, happy to share this weeks community update with Apache Flink 1.9, bylaws for Apache Flink, Savepoints vs Checkpoints, Flink on ARM, and more. As always, please feel free to add additional updates and news to this thread! Flink Development === * [releases] The release

Apache Flink - Manipulating the state of an object in an evictor or trigger

Hi: Is it safe to manipulate the state of an object in the evictor or trigger ?  Are there any best practices/dos and don't on this ? Thanks

?????? [flink 1.8.1]window closed unexpectedly and data drop

timestamp of the fourth data is (03:17:55), and the watermark time should be 03:17:50(water mark is 5). That time, window of the first data(ts:03:15:48) should be closed. What's more, there're so many sliding windows, and some of them should be closed to. --

State incompatible

Hi, I added a ttl to my state *old version :* private lazy val stateDescriptor = new ValueStateDescriptor("foo", Types.CASE_CLASS[DomainState]) *vs the new version * @transient private lazy val storeTtl = StateTtlConfig.newBuilder(90) .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)