Java SDK

Stateful functions are the building blocks of applications; they are atomic units of isolation, distribution, and persistence. As objects, they encapsulate the state of a single entity (e.g., a specific user, device, or session) and encode its behavior. Stateful functions can interact with each other, and external systems, through message passing. The Java SDK is supported as an embedded module.

To get started, add the Java SDK as a dependency to your application.

  1. <dependency>
  2.     <groupId>org.apache.flink</groupId>
  3.     <artifactId>statefun-sdk</artifactId>
  4.     <version>2.2.0</version>
  5. </dependency>

Defining A Stateful Function

A stateful function is any class that implements the StatefulFunction interface. The following is an example of a simple hello world function.

  1. package org.apache.flink.statefun.docs;
  3. import org.apache.flink.statefun.sdk.Context;
  4. import org.apache.flink.statefun.sdk.StatefulFunction;
  6. public class FnHelloWorld implements StatefulFunction {
  8.     @Override
  9.     public void invoke(Context context, Object input) {
  10.         System.out.println("Hello " + input.toString());
  11.     }
  12. }

Functions process each incoming message through their invoke method. Input’s are untyped and passed through the system as a java.lang.Object so one function can potentially process multiple types of messages.

The Context provides metadata about the current message and function, and is how you can call other functions or external systems. Functions are invoked based on a function type and unique identifier.

Stateful Match Function

Stateful functions provide a powerful abstraction for working with events and state, allowing developers to build components that can react to any kind of message. Commonly, functions only need to handle a known set of message types, and the StatefulMatchFunction interface provides an opinionated solution to that problem.

Simple Match Function

Stateful match functions are an opinionated variant of stateful functions for precisely this pattern. Developers outline expected types, optional predicates, and well-typed business logic and let the system dispatch each input to the correct action. Variants are bound inside a configure method that is executed once the first time an instance is loaded.

  1. package org.apache.flink.statefun.docs.match;
  3. import org.apache.flink.statefun.sdk.Context;
  4. import org.apache.flink.statefun.sdk.match.MatchBinder;
  5. import org.apache.flink.statefun.sdk.match.StatefulMatchFunction;
  7. public class FnMatchGreeter extends StatefulMatchFunction {
  9.     @Override
  10.     public void configure(MatchBinder binder) {
  11.         binder
  12.             .predicate(Customer.class, this::greetCustomer)
  13.             .predicate(Employee.class, Employee::isManager, this::greetManager)
  14.             .predicate(Employee.class, this::greetEmployee);
  15.     }
  17.     private void greetCustomer(Context context, Customer message) {
  18.         System.out.println("Hello customer " + message.getName());
  19.     }
  21.     private void greetEmployee(Context context, Employee message) {
  22.         System.out.println("Hello employee " + message.getEmployeeId());
  23.     }
  25.     private void greetManager(Context context, Employee message) {
  26.         System.out.println("Hello manager " + message.getEmployeeId());
  27.     }
  28. }

Making Your Function Complete

Similar to the first example, match functions are partial by default and will throw an IllegalStateException on any input that does not match any branch. They can be made complete by providing an otherwise clause that serves as a catch-all for unmatched input, think of it as a default clause in a Java switch statement. The otherwise action takes its message as an untyped java.lang.Object, allowing you to handle any unexpected messages.

  1. package org.apache.flink.statefun.docs.match;
  3. import org.apache.flink.statefun.sdk.Context;
  4. import org.apache.flink.statefun.sdk.match.MatchBinder;
  5. import org.apache.flink.statefun.sdk.match.StatefulMatchFunction;
  7. public class FnMatchGreeterWithCatchAll extends StatefulMatchFunction {
  9.     @Override
  10.     public void configure(MatchBinder binder) {
  11.         binder
  12.             .predicate(Customer.class, this::greetCustomer)
  13.             .predicate(Employee.class, Employee::isManager, this::greetManager)
  14.             .predicate(Employee.class, this::greetEmployee)
  15.             .otherwise(this::catchAll);
  16.     }
  18.     private void greetCustomer(Context context, Customer message) {
  19.         System.out.println("Hello customer " + message.getName());
  20.     }
  22.     private void greetEmployee(Context context, Employee message) {
  23.         System.out.println("Hello employee " + message.getEmployeeId());
  24.     }
  26.     private void greetManager(Context context, Employee message) {
  27.         System.out.println("Hello manager " + message.getEmployeeId());
  28.     }
  30.     private void catchAll(Context context, Object message) {
  31.         System.out.println("Hello unexpected message");
  32.     }
  33. }

Action Resolution Order

Match functions will always match actions from most to least specific using the following resolution rules.

First, find an action that matches the type and predicate. If two predicates will return true for a particular input, the one registered in the binder first wins. Next, search for an action that matches the type but does not have an associated predicate. Finally, if a catch-all exists, it will be executed or an IllegalStateException will be thrown.

Function Types and Messaging

In Java, function types are defined as logical pointers composed of a namespace and name. The type is bound to the implementing class in the module definition. Below is an example function type for the hello world function.

  1. package org.apache.flink.statefun.docs;
  3. import org.apache.flink.statefun.sdk.FunctionType;
  5. /** A function type that will be bound to {@link FnHelloWorld}. */
  6. public class Identifiers {
  8.   public static final FunctionType HELLO_TYPE = new FunctionType("apache/flink", "hello");
  9. }

This type can then be referenced from other functions to create an address and message a particular instance.

  1. package org.apache.flink.statefun.docs;
  3. import org.apache.flink.statefun.sdk.Context;
  4. import org.apache.flink.statefun.sdk.StatefulFunction;
  6. /** A simple stateful function that sends a message to the user with id "user1" */
  7. public class FnCaller implements StatefulFunction {
  9.   @Override
  10.   public void invoke(Context context, Object input) {
  11.     context.send(Identifiers.HELLO_TYPE, "user1", new MyUserMessage());
  12.   }
  13. }

Sending Delayed Messages

Functions are able to send messages on a delay so that they will arrive after some duration. Functions may even send themselves delayed messages that can serve as a callback. The delayed message is non-blocking so functions will continue to process records between the time a delayed message is sent and received.

  1. package org.apache.flink.statefun.docs.delay;
  3. import java.time.Duration;
  4. import org.apache.flink.statefun.sdk.Context;
  5. import org.apache.flink.statefun.sdk.StatefulFunction;
  7. public class FnDelayedMessage implements StatefulFunction {
  9.     @Override
  10.     public void invoke(Context context, Object input) {
  11.         if (input instanceof Message) {
  12.             System.out.println("Hello");
  13.             context.sendAfter(Duration.ofMinutes(1), context.self(), new DelayedMessage());
  14.         }
  16.         if (input instanceof DelayedMessage) {
  17.             System.out.println("Welcome to the future!");
  18.         }
  19.     }
  20. }

Completing Async Requests

When interacting with external systems, such as a database or API, one needs to take care that communication delay with the external system does not dominate the application’s total work. Stateful Functions allows registering a Java CompletableFuture that will resolve to a value at some point in the future. Future’s are registered along with a metadata object that provides additional context about the caller.

When the future completes, either successfully or exceptionally, the caller function type and id will be invoked with a AsyncOperationResult. An asynchronous result can complete in one of three states:

Success

The asynchronous operation has succeeded, and the produced result can be obtained via AsyncOperationResult#value.

Failure

The asynchronous operation has failed, and the cause can be obtained via AsyncOperationResult#throwable.

Unknown

The stateful function was restarted, possibly on a different machine, before the CompletableFuture was completed, therefore it is unknown what is the status of the asynchronous operation.

  1. package org.apache.flink.statefun.docs.async;
  3. import java.util.concurrent.CompletableFuture;
  4. import org.apache.flink.statefun.sdk.AsyncOperationResult;
  5. import org.apache.flink.statefun.sdk.Context;
  6. import org.apache.flink.statefun.sdk.StatefulFunction;
  8. @SuppressWarnings("unchecked")
  9. public class EnrichmentFunction implements StatefulFunction {
  11.     private final QueryService client;
  13.     public EnrichmentFunction(QueryService client) {
  14.         this.client = client;
  15.     }
  17.     @Override
  18.     public void invoke(Context context, Object input) {
  19.         if (input instanceof User) {
  20.             onUser(context, (User) input);
  21.         } else if (input instanceof AsyncOperationResult) {
  22.             onAsyncResult((AsyncOperationResult) input);
  23.         }
  24.     }
  26.     private void onUser(Context context, User user) {
  27.         CompletableFuture<UserEnrichment> future = client.getDataAsync(user.getUserId());
  28.         context.registerAsyncOperation(user, future);
  29.     }
  31.     private void onAsyncResult(AsyncOperationResult<User, UserEnrichment> result) {
  32.         if (result.successful()) {
  33.             User metadata = result.metadata();
  34.             UserEnrichment value = result.value();
  35.             System.out.println(
  36.                 String.format("Successfully completed future: %s %s", metadata, value));
  37.         } else if (result.failure()) {
  38.             System.out.println(
  39.                 String.format("Something has gone terribly wrong %s", result.throwable()));
  40.         } else {
  41.             System.out.println("Not sure what happened, maybe retry");
  42.         }
  43.     }
  44. }

Persistence

Stateful Functions treats state as a first class citizen and so all stateful functions can easily define state that is automatically made fault tolerant by the runtime. All stateful functions may contain state by merely defining one or more persisted fields.

The simplest way to get started is with a PersistedValue, which is defined by its name and the class of the type that it stores. The data is always scoped to a specific function type and identifier. Below is a stateful function that greets users based on the number of times they have been seen.

Attention: All PersistedValue, PersistedTable, and PersistedAppendingBuffer fields must be marked with a @Persisted annotation or they will not be made fault tolerant by the runtime.

  1. package org.apache.flink.statefun.docs;
  3. import org.apache.flink.statefun.sdk.Context;
  4. import org.apache.flink.statefun.sdk.FunctionType;
  5. import org.apache.flink.statefun.sdk.StatefulFunction;
  6. import org.apache.flink.statefun.sdk.annotations.Persisted;
  7. import org.apache.flink.statefun.sdk.state.PersistedValue;
  9. public class FnUserGreeter implements StatefulFunction {
  11.     public static FunctionType TYPE = new FunctionType("example", "greeter");
  13.     @Persisted
  14.     private final PersistedValue<Integer> count = PersistedValue.of("count", Integer.class);
  16.     public void invoke(Context context, Object input) {
  17.         String userId = context.self().id();
  18.         int seen = count.getOrDefault(0);
  20.         switch (seen) {
  21.             case 0:
  22.                 System.out.println(String.format("Hello %s!", userId));
  23.                 break;
  24.             case 1:
  25.                 System.out.println("Hello Again!");
  26.                 break;
  27.             case 2:
  28.                 System.out.println("Third time is the charm :)");
  29.                 break;
  30.             default:
  31.                 System.out.println(String.format("Hello for the %d-th time", seen + 1));
  32.         }
  34.         count.set(seen + 1);
  35.     }
  36. }

PersistedValue comes with the right primitive methods to build powerful stateful applications. Calling PersistedValue#get will return the current value of an object stored in state, or null if nothing is set. Conversely, PersistedValue#set will update the value in state and PersistedValue#clear will delete the value from state.

Collection Types

Along with PersistedValue, the Java SDK supports two persisted collection types. PersistedTable is a collection of keys and values, and PersistedAppendingBuffer is an append-only buffer.

These types are functionally equivalent to PersistedValue<Map> and PersistedValue<Collection> respectively but may provide better performance in some situations.

  1. @Persisted
  2. PersistedTable<String, Integer> table = PersistedTable.of("my-table", String.class, Integer.class);
  4. @Persisted
  5. PersistedAppendingBuffer<Integer> buffer = PersistedAppendingBuffer.of("my-buffer", Integer.class);

Dynamic State Registration

Using the above state types, a function’s persisted state must be defined eagerly. You cannot use those state types to register a new persisted state during invocations (i.e., in the invoke method) or after the function instance is created.

If dynamic state registration is required, it can be achieved using a PersistedStateRegistry:

  1. import org.apache.flink.statefun.sdk.Context;
  2. import org.apache.flink.statefun.sdk.FunctionType;
  3. import org.apache.flink.statefun.sdk.StatefulFunction;
  4. import org.apache.flink.statefun.sdk.annotations.Persisted;
  5. import org.apache.flink.statefun.sdk.state.PersistedStateRegistry;
  6. import org.apache.flink.statefun.sdk.state.PersistedValue;
  8. public class MyFunction implements StatefulFunction {
  10.     @Persisted
  11.     private final PersistedStateRegistry registry = new PersistedStateRegistry();
  13.     private PersistedValue<Integer> value;
  15.     public void invoke(Context context, Object input) {
  16.         if (value == null) {
  17.             value = PersistedValue.of("my-value", Integer.class);
  18.             registry.registerValue(valueOne);
  19.         }
  20.         int count = value.getOrDefault(0);
  21.         // ...
  22.     }
  23. }

Note how the PersistedValue field doesn’t need to be annotated with the @Persisted annotations, and is initially empty. The state object is dynamically created during invocation and registered with the PersistedStateRegistry so that the system picks it up to be managed for fault-tolerance.

State Expiration

Persisted states may be configured to expire and be deleted after a specified duration. This is supported by all types of state:

  1. @Persisted
  2. PersistedValue<Integer> table = PersistedValue.of(
  3.     "my-value",
  4.     Integer.class,
  5.     Expiration.expireAfterWriting(Duration.ofHours(1)));
  7. @Persisted
  8. PersistedTable<String, Integer> table = PersistedTable.of(
  9.     "my-table",
  10.     String.class,
  11.     Integer.class,
  12.     Expiration.expireAfterWriting(Duration.ofMinutes(5)));
  14. @Persisted
  15. PersistedAppendingBuffer<Integer> buffer = PersistedAppendingBuffer.of(
  16.     "my-buffer",
  17.     Integer.class,
  18.     Expiration.expireAfterWriting(Duration.ofSeconds(30)));

There are two expiration modes supported:

  1. Expiration.expireAfterWriting(...)
  3. Expiration.expireAfterReadingOrWriting(...)

State TTL configurations are made fault-tolerant by the runtime. In the case of downtime, state entries that should have been removed during said downtime will be purged immediately on restart.

Function Providers and Dependency Injection

Stateful functions are created across a distributed cluster of nodes. StatefulFunctionProvider is a factory class for creating a new instance of a StatefulFunction the first time it is activated.

  1. package org.apache.flink.statefun.docs;
  3. import org.apache.flink.statefun.docs.dependency.ProductionDependency;
  4. import org.apache.flink.statefun.docs.dependency.RuntimeDependency;
  5. import org.apache.flink.statefun.sdk.FunctionType;
  6. import org.apache.flink.statefun.sdk.StatefulFunction;
  7. import org.apache.flink.statefun.sdk.StatefulFunctionProvider;
  9. public class CustomProvider implements StatefulFunctionProvider {
  11.     public StatefulFunction functionOfType(FunctionType type) {
  12.         RuntimeDependency dependency = new ProductionDependency();
  13.         return new FnWithDependency(dependency);
  14.     }
  15. }

Providers are called once per type on each parallel worker, not for each id. If a stateful function requires custom configurations, they can be defined inside a provider and passed to the functions’ constructor. This is also where shared physical resources, such as a database connection, can be created that are used by any number of virtual functions. Now, tests can quickly provide mock, or test dependencies, without the need for complex dependency injection frameworks.

  1. package org.apache.flink.statefun.docs;
  3. import org.apache.flink.statefun.docs.dependency.RuntimeDependency;
  4. import org.apache.flink.statefun.docs.dependency.TestDependency;
  5. import org.junit.Assert;
  6. import org.junit.Test;
  8. public class FunctionTest {
  10.     @Test
  11.     public void testFunctionWithCustomDependency() {
  12.         RuntimeDependency dependency = new TestDependency();
  13.         FnWithDependency function = new FnWithDependency(dependency);
  15.         Assert.assertEquals("It appears math is broken", 1 + 1, 2);
  16.     }
  17. }