This is an automated email from the ASF dual-hosted git repository.
hxb pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 9be49ff871f [FLINK-27657][python] Implement remote operator state
backend in PyFlink
9be49ff871f is described below
commit 9be49ff871feace87aed9d4e3f8132bcf0cd3945
Author: Juntao Hu <[email protected]>
AuthorDate: Tue May 17 14:56:28 2022 +0800
[FLINK-27657][python] Implement remote operator state backend in PyFlink
This closes #19743.
---
flink-python/pyflink/datastream/__init__.py | 5 +
flink-python/pyflink/datastream/functions.py | 3 +-
flink-python/pyflink/datastream/state.py | 128 +++++++++++++++++++-
.../pyflink/fn_execution/internal_state.py | 10 +-
flink-python/pyflink/fn_execution/state_impl.py | 133 +++++++++++++++++++--
...eHandler.java => AbstractBeamStateHandler.java} | 22 +++-
.../python/beam/state/BeamBagStateHandler.java | 25 +---
.../python/beam/state/BeamMapStateHandler.java | 27 +----
.../python/beam/state/BeamStateHandler.java | 15 +++
9 files changed, 305 insertions(+), 63 deletions(-)
diff --git a/flink-python/pyflink/datastream/__init__.py
b/flink-python/pyflink/datastream/__init__.py
index 50c49352035..a9c21196ba0 100644
--- a/flink-python/pyflink/datastream/__init__.py
+++ b/flink-python/pyflink/datastream/__init__.py
@@ -141,6 +141,11 @@ Classes for state operations:
- :class:`state.AggregatingState`:
Interface for aggregating state, based on an :class:`AggregateFunction`.
Elements that are
added to this type of state will be eagerly pre-aggregated using a given
AggregateFunction.
+ - :class:`state.BroadcastState`:
+ A type of state that can be created to store the state of a
:class:`BroadcastStream`. This
+ state assumes that the same elements are sent to all instances of an
operator.
+ - :class:`state.ReadOnlyBroadcastState`:
+ A read-only view of the :class:`BroadcastState`.
- :class:`state.StateTtlConfig`:
Configuration of state TTL logic.
diff --git a/flink-python/pyflink/datastream/functions.py
b/flink-python/pyflink/datastream/functions.py
index c0c6738b141..9ee869154ac 100644
--- a/flink-python/pyflink/datastream/functions.py
+++ b/flink-python/pyflink/datastream/functions.py
@@ -17,9 +17,8 @@
################################################################################
from abc import ABC, abstractmethod
-from typing import Union, Any, Generic, TypeVar, Iterable
-
from py4j.java_gateway import JavaObject
+from typing import Union, Any, Generic, TypeVar, Iterable
from pyflink.datastream.state import ValueState, ValueStateDescriptor,
ListStateDescriptor, \
ListState, MapStateDescriptor, MapState, ReducingStateDescriptor,
ReducingState, \
diff --git a/flink-python/pyflink/datastream/state.py
b/flink-python/pyflink/datastream/state.py
index 05594df3822..b60bcdeb6ee 100644
--- a/flink-python/pyflink/datastream/state.py
+++ b/flink-python/pyflink/datastream/state.py
@@ -17,11 +17,10 @@
################################################################################
from abc import ABC, abstractmethod
from enum import Enum
-
from typing import TypeVar, Generic, Iterable, List, Iterator, Dict, Tuple,
Optional
-from pyflink.common.typeinfo import TypeInformation, Types
from pyflink.common.time import Time
+from pyflink.common.typeinfo import TypeInformation, Types
__all__ = [
'ValueStateDescriptor',
@@ -34,7 +33,10 @@ __all__ = [
'ReducingState',
'AggregatingStateDescriptor',
'AggregatingState',
- 'StateTtlConfig'
+ 'ReadOnlyBroadcastState',
+ 'BroadcastState',
+ 'StateTtlConfig',
+ 'OperatorStateStore',
]
T = TypeVar('T')
@@ -44,6 +46,22 @@ IN = TypeVar('IN')
OUT = TypeVar('OUT')
+class OperatorStateStore(ABC):
+ """
+ Interface for getting operator states. Currently, only
:class:`~state.BroadcastState` is
+ supported.
+ .. versionadded:: 1.16.0
+ """
+
+ @abstractmethod
+ def get_broadcast_state(self, state_descriptor: 'MapStateDescriptor') ->
'BroadcastState':
+ """
+ Fetches the :class:`~state.BroadcastState` described by
:class:`~state.MapStateDescriptor`,
+ which has read/write access to the broadcast operator state.
+ """
+ pass
+
+
class State(ABC):
"""
Interface that different types of partitioned state must implement.
@@ -281,6 +299,110 @@ class MapState(State, Generic[K, V]):
return iter(self.keys())
+class ReadOnlyBroadcastState(State, Generic[K, V]):
+ """
+ A read-only view of the :class:`BroadcastState`.
+ Although read-only, the user code should not modify the value returned by
the :meth:`get` or the
+ items returned by :meth:`items`, as this can lead to inconsistent states.
The reason for this is
+ that we do not create extra copies of the elements for performance reasons.
+ """
+
+ @abstractmethod
+ def get(self, key: K) -> V:
+ """
+ Returns the current value associated with the given key.
+ """
+ pass
+
+ @abstractmethod
+ def contains(self, key: K) -> bool:
+ """
+ Returns whether there exists the given mapping.
+ """
+ pass
+
+ @abstractmethod
+ def items(self) -> Iterable[Tuple[K, V]]:
+ """
+ Returns all the mappings in the state.
+ """
+ pass
+
+ @abstractmethod
+ def keys(self) -> Iterable[K]:
+ """
+ Returns all the keys in the state.
+ """
+ pass
+
+ @abstractmethod
+ def values(self) -> Iterable[V]:
+ """
+ Returns all the values in the state.
+ """
+ pass
+
+ @abstractmethod
+ def is_empty(self) -> bool:
+ """
+ Returns true if this state contains no key-value mappings, otherwise
false.
+ """
+ pass
+
+ def __getitem__(self, key: K) -> V:
+ return self.get(key)
+
+ def __contains__(self, key: K) -> bool:
+ return self.contains(key)
+
+ def __iter__(self) -> Iterator[K]:
+ return iter(self.keys())
+
+
+class BroadcastState(ReadOnlyBroadcastState[K, V]):
+ """
+ A type of state that can be created to store the state of a
:class:`BroadcastStream`. This state
+ assumes that the same elements are sent to all instances of an operator.
+ CAUTION: the user has to guarantee that all task instances store the same
elements in
+ this type of state.
+ Each operator instance individually maintains and stores elements in the
broadcast state. The
+ fact that the incoming stream is a broadcast one guarantees that all
instances see all the
+ elements. Upon recovery or re-scaling, the same state is given to each of
the instances.
+ To avoid hotspots, each task reads its previous partition, and if there
are more tasks (scale up
+ ), then the new instances read from the old instances in a round-robin
fashion. This is why each
+ instance has to guarantee that it stores the same elements as the rest. If
not, upon recovery or
+ rescaling you may have unpredictable redistribution of the partitions,
thus unpredictable
+ results.
+ """
+
+ @abstractmethod
+ def put(self, key: K, value: V) -> None:
+ """
+ Associates a new value with the given key.
+ """
+ pass
+
+ @abstractmethod
+ def put_all(self, dict_value: Dict[K, V]) -> None:
+ """
+ Copies all of the mappings from the given map into the state.
+ """
+ pass
+
+ @abstractmethod
+ def remove(self, key: K) -> None:
+ """
+ Deletes the mapping of the given key.
+ """
+ pass
+
+ def __setitem__(self, key: K, value: V) -> None:
+ self.put(key, value)
+
+ def __delitem__(self, key: K) -> None:
+ self.remove(key)
+
+
class StateDescriptor(ABC):
"""
Base class for state descriptors. A StateDescriptor is used for creating
partitioned State in
diff --git a/flink-python/pyflink/fn_execution/internal_state.py
b/flink-python/pyflink/fn_execution/internal_state.py
index 47512fd0129..5a0bd5627a6 100644
--- a/flink-python/pyflink/fn_execution/internal_state.py
+++ b/flink-python/pyflink/fn_execution/internal_state.py
@@ -19,7 +19,7 @@ from abc import ABC, abstractmethod
from typing import Generic, TypeVar, List, Iterable, Collection
from pyflink.datastream.state import State, ValueState, AppendingState,
MergingState, ListState, \
- AggregatingState, ReducingState, MapState
+ AggregatingState, ReducingState, MapState, ReadOnlyBroadcastState,
BroadcastState
N = TypeVar('N')
T = TypeVar('T')
@@ -111,3 +111,11 @@ class InternalMapState(InternalKvState[N], MapState[K, V],
ABC):
The peer to the :class:MapState in the internal state type hierarchy.
"""
pass
+
+
+class InternalReadOnlyBroadcastState(ReadOnlyBroadcastState[K, V], ABC):
+ pass
+
+
+class InternalBroadcastState(InternalReadOnlyBroadcastState[K, V],
BroadcastState[K, V], ABC):
+ pass
diff --git a/flink-python/pyflink/fn_execution/state_impl.py
b/flink-python/pyflink/fn_execution/state_impl.py
index 38668dd4e83..1252508ccfc 100644
--- a/flink-python/pyflink/fn_execution/state_impl.py
+++ b/flink-python/pyflink/fn_execution/state_impl.py
@@ -18,24 +18,24 @@
import base64
import collections
from abc import ABC, abstractmethod
-from enum import Enum
-from functools import partial
-from io import BytesIO
-
from apache_beam.coders import coder_impl
from apache_beam.portability.api import beam_fn_api_pb2
from apache_beam.runners.worker.bundle_processor import
SynchronousBagRuntimeState
from apache_beam.transforms import userstate
-from typing import List, Tuple, Any, Dict, Collection
+from enum import Enum
+from functools import partial
+from io import BytesIO
+from typing import List, Tuple, Any, Dict, Collection, cast
from pyflink.datastream import ReduceFunction
from pyflink.datastream.functions import AggregateFunction
-from pyflink.datastream.state import StateTtlConfig
+from pyflink.datastream.state import StateTtlConfig, MapStateDescriptor,
OperatorStateStore
from pyflink.fn_execution.beam.beam_coders import FlinkCoder
-from pyflink.fn_execution.coders import FieldCoder
+from pyflink.fn_execution.coders import FieldCoder, MapCoder, from_type_info
+from pyflink.fn_execution.flink_fn_execution_pb2 import StateDescriptor as
pb2_StateDescriptor
from pyflink.fn_execution.internal_state import InternalKvState, N,
InternalValueState, \
InternalListState, InternalReducingState, InternalMergingState,
InternalAggregatingState, \
- InternalMapState
+ InternalMapState, InternalReadOnlyBroadcastState, InternalBroadcastState
class LRUCache(object):
@@ -1094,8 +1094,7 @@ class RemoteKeyedStateBackend(object):
self, name, encoded_namespace, map_key_coder, map_value_coder,
ttl_config, cache_type):
# Currently the `beam_fn_api.proto` does not support MapState, so we
use the
# the `MultimapSideInput` message to mark the state as a MapState for
now.
- from pyflink.fn_execution.flink_fn_execution_pb2 import StateDescriptor
- state_proto = StateDescriptor()
+ state_proto = pb2_StateDescriptor()
state_proto.state_name = name
if ttl_config is not None:
state_proto.state_ttl_config.CopyFrom(ttl_config._to_proto())
@@ -1216,3 +1215,117 @@ class RemoteKeyedStateBackend(object):
if isinstance(internal_state, SynchronousBagRuntimeState):
internal_state._cleared = False
internal_state._added_elements = []
+
+
+class SynchronousReadOnlyBroadcastRuntimeState(InternalReadOnlyBroadcastState):
+ def __init__(self, name: str, internal_map_state:
"InternalSynchronousMapRuntimeState"):
+ self._name = name
+ self._internal_map_state = internal_map_state
+
+ def get(self, key):
+ return self._internal_map_state.get(key)
+
+ def contains(self, key) -> bool:
+ return self._internal_map_state.contains(key)
+
+ def items(self):
+ return self._internal_map_state.items()
+
+ def keys(self):
+ return self._internal_map_state.keys()
+
+ def values(self):
+ return self._internal_map_state.values()
+
+ def is_empty(self):
+ return self._internal_map_state.is_empty()
+
+ def clear(self):
+ return self._internal_map_state.clear()
+
+
+class SynchronousBroadcastRuntimeState(
+ SynchronousReadOnlyBroadcastRuntimeState, InternalBroadcastState
+):
+ def __init__(self, name: str, internal_map_state:
"InternalSynchronousMapRuntimeState"):
+ super(SynchronousBroadcastRuntimeState, self).__init__(name,
internal_map_state)
+
+ def put(self, key, value):
+ self._internal_map_state.put(key, value)
+
+ def put_all(self, dict_value):
+ self._internal_map_state.put_all(dict_value)
+
+ def remove(self, key):
+ self._internal_map_state.remove(key)
+
+ def commit(self):
+ self._internal_map_state.commit()
+
+ def to_read_only_broadcast_state(self) ->
"SynchronousReadOnlyBroadcastRuntimeState":
+ return SynchronousReadOnlyBroadcastRuntimeState(self._name,
self._internal_map_state)
+
+
+class RemoteOperatorStateBackend(OperatorStateStore):
+ def __init__(
+ self, state_handler, state_cache_size, map_state_read_cache_size,
map_state_write_cache_size
+ ):
+ self._state_handler = state_handler
+ self._state_cache_size = state_cache_size
+ # NOTE: if user stores a state into a class member, that state
actually won't be actually
+ # evicted from memory (because its counter > 0)
+ self._state_cache = LRUCache(state_cache_size, None)
+ self._state_cache.set_on_evict(lambda _, state: state.commit())
+ self._map_state_read_cache_size = map_state_read_cache_size
+ self._map_state_write_cache_size = map_state_write_cache_size
+ self._map_state_handler = CachingMapStateHandler(state_handler,
map_state_read_cache_size)
+
+ def get_broadcast_state(
+ self, state_descriptor: MapStateDescriptor
+ ) -> 'SynchronousBroadcastRuntimeState':
+ state_name = state_descriptor.name
+ map_coder = cast(MapCoder, from_type_info(state_descriptor.type_info))
# type: MapCoder
+ key_coder = map_coder._key_coder
+ value_coder = map_coder._value_coder
+
+ if state_name in self._state_cache:
+ self._validate_broadcast_state(state_name, key_coder, value_coder)
+ return self._state_cache.get(state_name)
+
+ state_proto = pb2_StateDescriptor()
+ state_proto.state_name = state_name
+ # Currently, MultimapKeysSideInput is used for BroadcastState
+ state_key = beam_fn_api_pb2.StateKey(
+
multimap_keys_side_input=beam_fn_api_pb2.StateKey.MultimapKeysSideInput(
+ transform_id="",
+ window=bytes(),
+
side_input_id=base64.b64encode(state_proto.SerializeToString()),
+ )
+ )
+
+ internal_map_state = InternalSynchronousMapRuntimeState(
+ self._map_state_handler,
+ state_key,
+ key_coder,
+ value_coder,
+ self._map_state_write_cache_size,
+ )
+
+ broadcast_state = SynchronousBroadcastRuntimeState(
+ state_descriptor.name, internal_map_state
+ )
+ self._state_cache.put(state_name, broadcast_state)
+ return broadcast_state
+
+ def commit(self):
+ for state in self._state_cache:
+ cast(SynchronousBroadcastRuntimeState, state).commit()
+
+ def _validate_broadcast_state(self, name, key_coder, value_coder):
+ if name in self._state_cache:
+ state = cast(SynchronousBroadcastRuntimeState,
self._state_cache.get(name))
+ if (
+ key_coder != state._internal_map_state._map_key_coder
+ or value_coder != state._internal_map_state._map_value_coder
+ ):
+ raise Exception("State name corrupted: %s" % name)
diff --git
a/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/state/BeamStateHandler.java
b/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/state/AbstractBeamStateHandler.java
similarity index 52%
copy from
flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/state/BeamStateHandler.java
copy to
flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/state/AbstractBeamStateHandler.java
index 90800c289bc..9f57de5f26f 100644
---
a/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/state/BeamStateHandler.java
+++
b/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/state/AbstractBeamStateHandler.java
@@ -20,9 +20,23 @@ package
org.apache.flink.streaming.api.runners.python.beam.state;
import org.apache.beam.model.fnexecution.v1.BeamFnApi;
-/** Interface for doing actual operations on Flink state based on {@link
BeamFnApi.StateRequest}. */
-public interface BeamStateHandler<S> {
+/** Abstract class extends {@link BeamStateHandler}, which implements the
common handle logic. */
+public abstract class AbstractBeamStateHandler<S> implements
BeamStateHandler<S> {
- BeamFnApi.StateResponse.Builder handle(BeamFnApi.StateRequest request, S
state)
- throws Exception;
+ public BeamFnApi.StateResponse.Builder handle(BeamFnApi.StateRequest
request, S state)
+ throws Exception {
+ switch (request.getRequestCase()) {
+ case GET:
+ return handleGet(request, state);
+ case APPEND:
+ return handleAppend(request, state);
+ case CLEAR:
+ return handleClear(request, state);
+ default:
+ throw new RuntimeException(
+ String.format(
+ "Unsupported request type %s for user state.",
+ request.getRequestCase()));
+ }
+ }
}
diff --git
a/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/state/BeamBagStateHandler.java
b/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/state/BeamBagStateHandler.java
index 08051a544a3..81956f048ff 100644
---
a/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/state/BeamBagStateHandler.java
+++
b/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/state/BeamBagStateHandler.java
@@ -36,7 +36,7 @@ import java.util.List;
import java.util.Set;
/** BeamBagStateHandler handles operations on {@link ListState}, which backs
Beam bag states. */
-public class BeamBagStateHandler implements
BeamStateHandler<ListState<byte[]>> {
+public class BeamBagStateHandler extends
AbstractBeamStateHandler<ListState<byte[]>> {
private static final String MERGE_NAMESPACES_MARK = "merge_namespaces";
@@ -54,24 +54,7 @@ public class BeamBagStateHandler implements
BeamStateHandler<ListState<byte[]>>
this.baisWrapper = new DataInputViewStreamWrapper(bais);
}
- public BeamFnApi.StateResponse.Builder handle(
- BeamFnApi.StateRequest request, ListState<byte[]> listState)
throws Exception {
- switch (request.getRequestCase()) {
- case GET:
- return handleBagGetRequest(request, listState);
- case APPEND:
- return handleBagAppendRequest(request, listState);
- case CLEAR:
- return handleBagClearRequest(request, listState);
- default:
- throw new RuntimeException(
- String.format(
- "Unsupported request type %s for user state.",
- request.getRequestCase()));
- }
- }
-
- private BeamFnApi.StateResponse.Builder handleBagGetRequest(
+ public BeamFnApi.StateResponse.Builder handleGet(
BeamFnApi.StateRequest request, ListState<byte[]> listState)
throws Exception {
List<ByteString> byteStrings = convertToByteString(listState);
@@ -82,7 +65,7 @@ public class BeamBagStateHandler implements
BeamStateHandler<ListState<byte[]>>
.setData(ByteString.copyFrom(byteStrings)));
}
- private BeamFnApi.StateResponse.Builder handleBagAppendRequest(
+ public BeamFnApi.StateResponse.Builder handleAppend(
BeamFnApi.StateRequest request, ListState<byte[]> listState)
throws Exception {
if (request.getStateKey()
.getBagUserState()
@@ -113,7 +96,7 @@ public class BeamBagStateHandler implements
BeamStateHandler<ListState<byte[]>>
.setAppend(BeamFnApi.StateAppendResponse.getDefaultInstance());
}
- private BeamFnApi.StateResponse.Builder handleBagClearRequest(
+ public BeamFnApi.StateResponse.Builder handleClear(
BeamFnApi.StateRequest request, ListState<byte[]> listState)
throws Exception {
listState.clear();
diff --git
a/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/state/BeamMapStateHandler.java
b/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/state/BeamMapStateHandler.java
index 11e4810d094..cb055d059bd 100644
---
a/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/state/BeamMapStateHandler.java
+++
b/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/state/BeamMapStateHandler.java
@@ -36,7 +36,8 @@ import java.util.Map;
import java.util.UUID;
/** BeamMapStateHandler handles operations on a {@link MapState}. */
-public class BeamMapStateHandler implements
BeamStateHandler<MapState<ByteArrayWrapper, byte[]>> {
+public class BeamMapStateHandler
+ extends AbstractBeamStateHandler<MapState<ByteArrayWrapper, byte[]>> {
private static final String CLEAR_CACHED_ITERATOR_MARK = "clear_iterators";
@@ -110,25 +111,7 @@ public class BeamMapStateHandler implements
BeamStateHandler<MapState<ByteArrayW
}
}
- public BeamFnApi.StateResponse.Builder handle(
- BeamFnApi.StateRequest request, MapState<ByteArrayWrapper, byte[]>
mapState)
- throws Exception {
- switch (request.getRequestCase()) {
- case GET:
- return handleGet(request, mapState);
- case APPEND:
- return handleAppend(request, mapState);
- case CLEAR:
- return handleClear(request, mapState);
- default:
- throw new RuntimeException(
- String.format(
- "Unsupported request type %s for user state.",
- request.getRequestCase()));
- }
- }
-
- private BeamFnApi.StateResponse.Builder handleGet(
+ public BeamFnApi.StateResponse.Builder handleGet(
BeamFnApi.StateRequest request, MapState<ByteArrayWrapper, byte[]>
mapState)
throws Exception {
// The continuation token structure of GET request is:
@@ -174,7 +157,7 @@ public class BeamMapStateHandler implements
BeamStateHandler<MapState<ByteArrayW
return
BeamFnApi.StateResponse.newBuilder().setId(request.getId()).setGet(response);
}
- private BeamFnApi.StateResponse.Builder handleAppend(
+ public BeamFnApi.StateResponse.Builder handleAppend(
BeamFnApi.StateRequest request, MapState<ByteArrayWrapper, byte[]>
mapState)
throws Exception {
// The structure of append request bytes is:
@@ -219,7 +202,7 @@ public class BeamMapStateHandler implements
BeamStateHandler<MapState<ByteArrayW
.setAppend(BeamFnApi.StateAppendResponse.getDefaultInstance());
}
- private BeamFnApi.StateResponse.Builder handleClear(
+ public BeamFnApi.StateResponse.Builder handleClear(
BeamFnApi.StateRequest request, MapState<ByteArrayWrapper, byte[]>
mapState)
throws Exception {
if (request.getStateKey()
diff --git
a/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/state/BeamStateHandler.java
b/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/state/BeamStateHandler.java
index 90800c289bc..bb834f81e03 100644
---
a/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/state/BeamStateHandler.java
+++
b/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/state/BeamStateHandler.java
@@ -23,6 +23,21 @@ import org.apache.beam.model.fnexecution.v1.BeamFnApi;
/** Interface for doing actual operations on Flink state based on {@link
BeamFnApi.StateRequest}. */
public interface BeamStateHandler<S> {
+ /**
+ * Dispatches {@link BeamFnApi.StateRequest} to different handle functions
base on request case.
+ */
BeamFnApi.StateResponse.Builder handle(BeamFnApi.StateRequest request, S
state)
throws Exception;
+
+ /** Handles GET requests. */
+ BeamFnApi.StateResponse.Builder handleGet(BeamFnApi.StateRequest request,
S state)
+ throws Exception;
+
+ /** Handles APPEND requests. */
+ BeamFnApi.StateResponse.Builder handleAppend(BeamFnApi.StateRequest
request, S state)
+ throws Exception;
+
+ /** Handles CLEAR requests. */
+ BeamFnApi.StateResponse.Builder handleClear(BeamFnApi.StateRequest
request, S state)
+ throws Exception;
}
