Hello, I am try to wrap my head around the Flink Memory and concurrency model (including state), I have read (a few times) the "Flink architecture" and "Working withState" pages on the Flink website, but I am still a bit confused. From this and a few stack Overflow posts/answers, I have made up the following mental model • each "subtask" is an operator chain and is run by a single thread • the class instances that are corresponding to that operator therefore don't have concurrent access, e.g: on a specific

RichFlatMap

instance, there can not be a concurrent access to

flatMap()

. • different Task Slots belonging to the same Task Manager may hold analogous Task Chains. The operators running Those are spawned via serialization/deserialization from the ones Constructed in the

main()

"execution plan" and then initialized individually via

open()

for example. • Therefore, holding

static

attributes in the operator classes is prone to race conditions, but having non-static atttributes is not. However, they need to be flagged as `transient`and properly initialized in

open()

. However, those attributes will not be part of state and if I want them to be, I will need to convert those attributes into

ValueState<>

ones and manage them Is this mental model flawed? If yes, how?

sounds about right

@Chesnay Schepler Thanks for confirming! Until now, every time I thought I started to understand the way things were laid out, I stumbled on a new concept. Setting a working chain is deceiptfuly simple, it just works, but then the tricky question slowly creep up

I hope it’s not right because then my model is not right. • Subtasks are run in slots. Slots reserve statically an amout of memory. If I have a pipeline of 10 plus 10 tasks in two stages, it needs a total of 10 slots because by default they can share slots. • I don’t see that many threads running in prod, the load is a lot less.

What do you define as a "task"? It's not an exactly well-defined term and can be interpreted in different ways. Let's say you have a job like

chain(source -> map) -> sink

These are 2 vertices. If you run this with a parallelism of 1, then there are 2 subtasks total; one for the

chain(source -> map)

vertex, one for the

sink

. Both are deployed as a separate task (each using a different thread), but using 1 slot because of slot-sharing (by default!).