tessl/maven-org-apache-flink--flink-queryable-state-client-java

Java client library for accessing queryable state in Apache Flink streaming applications through a network-based query interface.

—

Pending

Overview

Eval results

Files

State Querying

Name: tessl/maven-org-apache-flink--flink-queryable-state-client-java
Author: tessl

Asynchronous state querying capabilities for accessing different types of keyed state from running Flink jobs. All state queries return immutable, read-only state objects.

Capabilities

Basic State Query

Query keyed state using TypeInformation for type safety.

public <K, S extends State, V> CompletableFuture<S> getKvState(
    JobID jobId,
    String queryableStateName,
    K key,
    TypeInformation<K> keyTypeInfo,
    StateDescriptor<S, V> stateDescriptor
);

Type Parameters:

K - Type of the key
S - Type of the state (extends State)
V - Type of the value stored in the state

Parameters:

jobId - JobID of the job containing the queryable state
queryableStateName - Name under which the state is queryable (set via StateDescriptor.setQueryable())
key - The key to query for
keyTypeInfo - TypeInformation for the key type
stateDescriptor - StateDescriptor matching the state configuration in the Flink job

Returns:

CompletableFuture<S> - Future containing an immutable state object

Usage Example:

import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;

QueryableStateClient client = new QueryableStateClient("localhost", 9069);

JobID jobId = JobID.fromHexString("a1b2c3d4e5f6789012345678901234567890abcd");
String stateName = "userProfiles";
String userId = "user123";

ValueStateDescriptor<UserProfile> descriptor = 
    new ValueStateDescriptor<>("userProfiles", UserProfile.class);

CompletableFuture<ValueState<UserProfile>> future = client.getKvState(
    jobId,
    stateName,
    userId,
    TypeInformation.of(String.class),
    descriptor
);

future.thenAccept(state -> {
    UserProfile profile = state.value();
    System.out.println("User profile: " + profile);
});

Type Hint-Based Query

Query keyed state using TypeHint for generic type inference.

public <K, S extends State, V> CompletableFuture<S> getKvState(
    JobID jobId,
    String queryableStateName,
    K key,
    TypeHint<K> keyTypeHint,
    StateDescriptor<S, V> stateDescriptor
);

Parameters:

Same as basic query, but uses TypeHint<K> instead of TypeInformation<K>

Usage Example:

import org.apache.flink.api.common.typeinfo.TypeHint;

CompletableFuture<ValueState<String>> future = client.getKvState(
    jobId,
    stateName,
    "myKey",
    new TypeHint<String>() {}, // Type hint for String key
    new ValueStateDescriptor<>("state", String.class)
);

Supported State Types

The client supports querying all Flink state types:

ValueState

ValueStateDescriptor<String> descriptor = 
    new ValueStateDescriptor<>("myValue", String.class);
CompletableFuture<ValueState<String>> future = client.getKvState(/*...*/);

ListState

ListStateDescriptor<Integer> descriptor = 
    new ListStateDescriptor<>("myList", Integer.class);
CompletableFuture<ListState<Integer>> future = client.getKvState(/*...*/);

MapState

MapStateDescriptor<String, Long> descriptor = 
    new MapStateDescriptor<>("myMap", String.class, Long.class);
CompletableFuture<MapState<String, Long>> future = client.getKvState(/*...*/);

ReducingState

ReducingStateDescriptor<Integer> descriptor = 
    new ReducingStateDescriptor<>("myReducing", Integer::sum, Integer.class);
CompletableFuture<ReducingState<Integer>> future = client.getKvState(/*...*/);

AggregatingState

AggregatingStateDescriptor<Integer, Long, Double> descriptor = 
    new AggregatingStateDescriptor<>("myAggregating", aggregateFunction, Long.class);
CompletableFuture<AggregatingState<Integer, Double>> future = client.getKvState(/*...*/);

Query Process

The state querying process involves several steps:

Key Serialization: The key and namespace are serialized using the provided type serializers
Location Resolution: The client proxy contacts the JobManager to resolve the location of the key group
State Request: A KvStateRequest is sent to the appropriate TaskManager
State Retrieval: The TaskManager retrieves the serialized state data
Response Processing: The serialized state is deserialized into an immutable state wrapper
Future Completion: The CompletableFuture is completed with the immutable state object

Error Handling

State queries can fail for various reasons. Common exceptions include:

UnknownKeyOrNamespaceException - No state exists for the given key/namespace
UnknownKvStateIdException - The state ID is not recognized
UnknownLocationException - State location cannot be resolved
IOException - Network or serialization errors
FlinkRuntimeException - General runtime errors

Error Handling Example:

future.whenComplete((state, throwable) -> {
    if (throwable != null) {
        if (throwable instanceof UnknownKeyOrNamespaceException) {
            System.out.println("No state found for key: " + key);
        } else if (throwable instanceof UnknownLocationException) {
            System.err.println("Could not resolve state location");
        } else {
            System.err.println("Query failed: " + throwable.getMessage());
        }
    } else {
        // Process successful result
        processState(state);
    }
});

Performance Considerations

Location Caching: Resolved state locations are cached to improve performance
Asynchronous Operations: All queries are non-blocking and return CompletableFuture
Serialization Overhead: Consider the cost of key and state serialization/deserialization
Network Communication: Each query involves network communication with the Flink cluster
Connection Reuse: The client maintains persistent connections to minimize connection overhead

Install with Tessl CLI

npx tessl i tessl/maven-org-apache-flink--flink-queryable-state-client-java

docs

client-management.md

exception-handling.md