private static class StatefulFunction implements OneInputStreamProcessFunction<Long, Long> {
    static final StateDeclaration.ListStateDeclaration<Long> LIST_STATE_DECLARATION =
            StateDeclarations.listStateBuilder("example-list-state", Types.LONG).redistributeBy(RedistributionStrategy.SPLIT).build();
    
	@Override
    public Set<StateDeclaration> usesStates() {
        return Collections.singleton(LIST_STATE_DECLARATION);
    } 

    @Override
    public void processRecord(Long record, Collector<Long> output, RuntimeContext ctx)
            throws Exception {
        Optional<ListState<Long>> stateOptional =
                ctx.getStateManager().getState(LIST_STATE_DECLARATION);
        ListState<Long> state = stateOptional.get();
        // do something with this state. For example, update the state by this record.
        state.update(Collections.singletonList(record));
    }
}

/** {@link StateDeclaration} represents a declaration of the specific state used. */
public interface StateDeclaration extends Serializable {

    /** Get the name of this state. */
    String getName();

    /**
     * Get the {@link RedistributionMode} of this state. More details see the doc of {@link
     * RedistributionMode}.
     */
    RedistributionMode getRedistributionMode();

    /**
     * {@link RedistributionMode} is used to indicate whether this state supports redistribution
     * between partitions and how to redistribute this state during rescaling.
     */
    enum RedistributionMode {
        /**
         * Not supports redistribution.
         *
         * <p>For example : KeyedState is bind to a specific keyGroup, so it is can't support
         * redistribution between partitions.
         */
        NONE,

        /**
         * This state can be safely redistributed between different partitions, and the specific
         * redistribution strategy is determined by the state itself.
         *
         * <p>For example: ListState's redistribution algorithm is determined by {@link
         * ListStateDeclaration.RedistributionStrategy}.
         */
        REDISTRIBUTABLE,

        /** States are guranteed to be identical in different partitions, thus redistribution is not a problem. */
        IDENTICAL
    }
}

/** This represents a declaration of the value state. */
public interface ValueStateDeclaration<T> extends StateDeclaration {
    /** Get type descriptor of this value state. */
    TypeDescriptor<T> getTypeDescriptor();
}

/** This represents a declaration of the list state. */
public interface ListStateDeclaration<T> extends StateDeclaration {
    /**
     * Get the {@link RedistributionStrategy} of this list state.
     *
     * @return the redistribution strategy of this list state.
     */
    RedistributionStrategy getRedistributionStrategy();

    /**
     * {@link RedistributionStrategy} is used to guide the assignment of states during rescaling.
     */
    enum RedistributionStrategy {
        /**
         * The whole state is logically a concatenation of all lists. On restore/redistribution, the
         * list is evenly divided into as many sub-lists as there are parallel operators. Each
         * operator gets a sub-list, which can be empty, or contain one or more elements.
         */
        SPLIT,
        /**
         * The whole state is logically a concatenation of all lists. On restore/redistribution,
         * each operator gets the complete list of state elements.
         */
        UNION
    }

    /** Get type descriptor of this list state's element. */
    TypeDescriptor<T> getTypeDescriptor();
}

/** This represents a declaration of the map state. */
public interface MapStateDeclaration<K, V> extends StateDeclaration {
    /** Get type descriptor of this map state's key. */
    TypeDescriptor<K> getKeyTypeDescriptor();

    /** Get type descriptor of this map state's value. */
    TypeDescriptor<V> getValueTypeDescriptor();
}

/** This represents a declaration of the reducing state. */
public interface ReducingStateDeclaration<T> extends StateDeclaration {
    /** Get type descriptor of this reducing state. */
    TypeDescriptor<T> getTypeDescriptor();
}

/** This represents a declaration of the aggregating state. */
public interface AggregatingStateDeclaration<IN, ACC, OUT> extends StateDeclaration {
    /** Get type descriptor of this aggregating state's input. */
    TypeDescriptor<IN> getInTypeDescriptor();

    /** Get type descriptor of this aggregating state's output. */
    TypeDescriptor<OUT> getOutTypeDescriptor();

    /** Get the aggregate function of this state. */
    AggregateFunction<IN, ACC, OUT> getAggregateFunction();
}

/** This represents a declaration of the broadcast state. */
public interface BroadcastStateDeclaration<K, V> extends StateDeclaration {
    /** Get type descriptor of this broadcast state's key. */
    TypeDescriptor<K> getKeyTypeDescriptor();

    /** Get type descriptor of this broadcast state's value. */
    TypeDescriptor<V> getValueTypeDescriptor();
}

/** This is a helper class for declaring various states. */
public class StateDeclarations {
    /** Get the builder of {@link ListStateDeclaration}. */
    public static <T> ListStateDeclarationBuilder<T> listStateBuilder(
            String name, TypeDescriptor<T> elementTypeDescriptor)

    /** Get the builder of {@link MapStateDeclaration}. */
    public static <K, V> MapStateDeclarationBuilder<K, V> mapStateBuilder(
            String name,
            TypeDescriptor<K> keyTypeDescriptor,
            TypeDescriptor<V> valueTypeDescriptor)

    /** Get the builder of {@link ValueStateDeclaration}. */
    public <T> ValueStateDeclarationBuilder<T> valueStateBuilder(
            String name, TypeDescriptor<T> valueType)

    /** Get the builder of {@link ReducingStateDeclaration}. */
    public <T> ReducingStateDeclarationBuilder<T> reducingStateBuilder(
            String name, TypeDescriptor<T> typeDescriptor)

    /** Get the builder of {@link AggregatingStateDeclaration}. */
    public <IN, OUT, ACC> AggregatingStateDeclarationBuilder<IN, OUT, ACC> aggregatingStateBuilder(
            String name, AggregateFunction<IN, ACC, OUT> aggregateFunction)

    /**
     * Get the {@link ListStateDeclaration} of list state with {@link RedistributionMode#NONE}. If
     * you want to configure it more elaborately, use {@link #listStateBuilder(String,
     * TypeDescriptor)}.
     */
    public static <T> ListStateDeclaration<T> listState(
            String name, TypeDescriptor<T> elementTypeDescriptor)

    /**
     * Get the {@link MapStateDeclaration} of map state with {@link RedistributionMode#NONE}. If you
     * want to configure it more elaborately, use {@link #mapStateBuilder(String, TypeDescriptor,
     * TypeDescriptor)}.
     */
    public static <K, V> MapStateDeclaration<K, V> mapState(
            String name,
            TypeDescriptor<K> keyTypeDescriptor,
            TypeDescriptor<V> valueTypeDescriptor)      
    
    /**
     * Get the {@link ValueStateDeclaration} of value state. If you want to configure it more
     * elaborately, use {@link #valueStateBuilder(String, TypeDescriptor)}.
     */
    public static <T> ValueStateDeclaration<T> valueState(String name, TypeDescriptor<T> valueType)

    /**
     * Get the {@link ReducingStateDeclaration} of list state. If you want to configure it more
     * elaborately, use {@link #reducingStateBuilder(String, TypeDescriptor)}.
     */
    public static <T> ReducingStateDeclaration<T> reducingState(
            String name, TypeDescriptor<T> typeDescriptor)

    /**
     * Get the {@link AggregatingStateDeclaration} of aggregating state. If you want to configure it
     * more elaborately, use {@link #aggregatingStateBuilder(String, AggregateFunction)}.
     */
    public static <IN, OUT, ACC> AggregatingStateDeclaration<IN, OUT, ACC> aggregatingState(
            String name, AggregateFunction<IN, ACC, OUT> aggregateFunction) 

    /** Builder for {@link ListStateDeclaration}. */
    public static class ListStateDeclarationBuilder<T> {

        ListStateDeclarationBuilder<T> redistributeBy(RedistributionStrategy strategy)

        ListStateDeclaration<T> build()
    }

    /** Builder for {@link MapStateDeclaration}. */
    public static class MapStateDeclarationBuilder<K, V> {
        MapStateDeclarationBuilder<K, V> broadcast()

        MapStateDeclaration<K, V> build()
    }

    /** Builder for {@link ValueStateDeclaration}. */
    public static class ValueStateDeclarationBuilder<T> {
        ValueStateDeclaration<T> build()
    }

    /** Builder for {@link ReducingStateDeclaration}. */
    public static class ReducingStateDeclarationBuilder<T> {
        ReducingStateDeclaration<T> build()
    }

    /** Builder for {@link AggregatingStateDeclaration}. */
    public static class AggregatingStateDeclarationBuilder<IN, OUT, ACC> {
        AggregatingStateDeclaration<IN, OUT, ACC> build()
    }
}

/**
 * Explicitly declares states upfront.Each specific state must be declared in this method before
 * it can be used.
 *
 * @return all declared states used by this process function.
 */
default Set<StateDeclaration> usesStates() {
    return Collections.emptySet();
}

    /**
     * Get the specific list state.
     *
     * @param stateDeclaration of this state.
     * @return the list state corresponds to the state declaration.
     */
    <T> Optional<ListState<T>> getState(ListStateDeclaration<T> stateDeclaration) throws Exception;

    /**
     * Get the specific value state.
     *
     * @param stateDeclaration of this state.
     * @return the value state corresponds to the state declaration.
     */
    <T> Optional<ValueState<T>> getState(ValueStateDeclaration<T> stateDeclaration)
            throws Exception;

    /**
     * Get the specific map state.
     *
     * @param stateDeclaration of this state.
     * @return the map state corresponds to the state declaration.
     */
    <K, V> Optional<MapState<K, V>> getState(MapStateDeclaration<K, V> stateDeclaration)
            throws Exception;

    /**
     * Get the specific reducing state.
     *
     * @param stateDeclaration of this state.
     * @return the reducing state corresponds to the state declaration.
     */
    <T> Optional<ReducingState<T>> getState(ReducingStateDeclaration<T> stateDeclaration)
            throws Exception;

    /**
     * Get the specific aggregating state.
     *
     * @param stateDeclaration of this state.
     * @return the aggregating state corresponds to the state declaration.
     */
    <IN, ACC, OUT> Optional<AggregatingState<IN, OUT>> getState(
            AggregatingStateDeclaration<IN, ACC, OUT> stateDeclaration) throws Exception;

    /**
     * Get the specific broadcast state.
     *
     * @param stateDeclaration of this state.
     * @return the broadcast state corresponds to the state declaration.
     */
    <K, V> Optional<BroadcastState<K, V>> getState(BroadcastStateDeclaration<K, V> stateDeclaration)
            throws Exception;