Flink Stateful functions - how state is scoped?

Question

I am going through the documentation on Flink's stateful functions, and got confused by several points.

Let's take the following example Java snippet (first one from this page: https://nightlies.apache.org/flink/flink-statefun-docs-release-3.2/docs/sdk/java/#serving-functions):

public class GreeterFn implements StatefulFunction {

    static final TypeName TYPE = TypeName.typeNameFromString("com.example.fns/greeter");

    static final TypeName INBOX = TypeName.typeNameFromString("com.example.fns/inbox");

    static final ValueSpec SEEN = ValueSpec.named("seen").withIntType();

    @Override 
    CompletableFuture apply(Context context, Message message) {
        String name = message.asUtf8String();

        var storage = context.storage();
        var seen = storage.get(SEEN).orElse(0);
        storage.set(SEEN, seen + 1);

        context.send(
            MessageBuilder.forAddress(INBOX, name)
                .withValue("Hello " + name + " for the " + seen + "th time!")
                .build());

        return context.done();
    }
}

The explanation under this snippet states the following:

This code declares a greeter function that will be registered under the logical type name "com.example.fns/greeter"

Q1. So does this mean, by convention, the stateful function class is always supposed to have a static TypeName variable, exactly called TYPE? and the value of this TYPE variable (i.e. namespace/name) is going to be used to send messages TO this stateful function?

Q2. Next, INBOX TypeName variable is used to specify a destination function to send a greeting message to, so as far as I understood, it does not necessarily need to be called INBOX, it could be any variable name?

Q3. Finally, it says:

It contains a single ValueSpec, which is implicitly scoped to the current address and stores an integer

I couldn't understand what's the scope of this ValueSpec and what is meant by "current address"? is it scoped by Message message received in apply()? let's say, if message.asUtf8String() is user1, is it going to have its own scope and ValueSpec, different from user2, user3, etc.? For example in Flink's KeyedProcessFunction, the state is scoped by the key, so does Message message in stateful function play a similar role of the key in KeyedProcessFunction?

Q4. Assuming above, when MessageBuilder.forAddress(INBOX, name) is called, does it mean that the apply() method of the stateful function with TYPE = com.example.fns/inbox will be called, and Message message is going to be name that we pass to forAddress()? What if more than one stateful function has that TYPE value?

Flink Stateful functions - how state is scoped?

0 Answers