You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by za...@apache.org on 2024/09/19 11:02:15 UTC
(flink) 01/02: [FLINK-36117] Setup internal interface hierarchy for state v2
This is an automated email from the ASF dual-hosted git repository.
zakelly pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 433794b565d5fe2a13f816a336205d4644903792
Author: Zakelly <za...@gmail.com>
AuthorDate: Fri Sep 13 18:07:28 2024 +0800
[FLINK-36117] Setup internal interface hierarchy for state v2
---
.../asyncprocessing/AsyncExecutionController.java | 2 +-
.../runtime/asyncprocessing/RecordContext.java | 2 +-
.../runtime/asyncprocessing/StateRequest.java | 2 +-
.../asyncprocessing/StateRequestHandler.java | 2 +-
...ingState.java => AbstractAggregatingState.java} | 21 ++++++++++++---
...rnalKeyedState.java => AbstractKeyedState.java} | 11 ++++----
...ternalListState.java => AbstractListState.java} | 7 ++---
...InternalMapState.java => AbstractMapState.java} | 7 ++---
...ducingState.java => AbstractReducingState.java} | 7 ++---
...rnalValueState.java => AbstractValueState.java} | 7 ++---
.../InternalAggregatingState.java} | 29 ++++++++-------------
.../InternalAppendingState.java} | 30 ++++++++--------------
.../InternalKeyedState.java} | 20 +++++++--------
.../InternalListState.java} | 22 ++++++++--------
.../InternalMapState.java} | 24 ++++++++---------
.../v2/{ => internal}/InternalMergingState.java | 20 ++++++++++++---
.../{ => internal}/InternalPartitionedState.java | 5 ++--
.../InternalReducingState.java} | 27 ++++++-------------
.../InternalValueState.java} | 22 ++++++++--------
.../AsyncExecutionControllerTest.java | 4 +--
...Test.java => AbstractAggregatingStateTest.java} | 8 +++---
...stBase.java => AbstractKeyedStateTestBase.java} | 4 +--
...stStateTest.java => AbstractListStateTest.java} | 8 +++---
...MapStateTest.java => AbstractMapStateTest.java} | 8 +++---
...ateTest.java => AbstractReducingStateTest.java} | 12 ++++-----
...eStateTest.java => AbstractValueStateTest.java} | 8 +++---
.../apache/flink/state/forst/ForStListState.java | 6 ++---
.../apache/flink/state/forst/ForStMapState.java | 6 ++---
.../flink/state/forst/ForStReducingState.java | 6 ++---
.../apache/flink/state/forst/ForStValueState.java | 6 ++---
.../state/forst/ForStDBIterateOperationTest.java | 2 +-
.../state/forst/ForStDBOperationTestBase.java | 2 +-
.../flink/state/forst/ForStStateExecutorTest.java | 4 +--
33 files changed, 174 insertions(+), 177 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java b/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java
index ad767aa340a..fc2d23c176d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java
@@ -25,7 +25,7 @@ import org.apache.flink.core.state.InternalStateFuture;
import org.apache.flink.core.state.StateFutureImpl.AsyncFrameworkExceptionHandler;
import org.apache.flink.runtime.asyncprocessing.EpochManager.ParallelMode;
import org.apache.flink.runtime.state.KeyGroupRangeAssignment;
-import org.apache.flink.runtime.state.v2.InternalPartitionedState;
+import org.apache.flink.runtime.state.v2.internal.InternalPartitionedState;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.function.ThrowingRunnable;
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/RecordContext.java b/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/RecordContext.java
index b9f48125c29..1dcd4761955 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/RecordContext.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/RecordContext.java
@@ -19,7 +19,7 @@
package org.apache.flink.runtime.asyncprocessing;
import org.apache.flink.runtime.asyncprocessing.EpochManager.Epoch;
-import org.apache.flink.runtime.state.v2.InternalPartitionedState;
+import org.apache.flink.runtime.state.v2.internal.InternalPartitionedState;
import javax.annotation.Nullable;
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequest.java b/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequest.java
index 32527a2e881..dea08422de6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequest.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequest.java
@@ -20,7 +20,7 @@ package org.apache.flink.runtime.asyncprocessing;
import org.apache.flink.api.common.state.v2.State;
import org.apache.flink.core.state.InternalStateFuture;
-import org.apache.flink.runtime.state.v2.InternalPartitionedState;
+import org.apache.flink.runtime.state.v2.internal.InternalPartitionedState;
import javax.annotation.Nullable;
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequestHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequestHandler.java
index e1ab3a7dda3..96873fc16f1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequestHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequestHandler.java
@@ -21,7 +21,7 @@ package org.apache.flink.runtime.asyncprocessing;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.state.v2.State;
import org.apache.flink.core.state.InternalStateFuture;
-import org.apache.flink.runtime.state.v2.InternalPartitionedState;
+import org.apache.flink.runtime.state.v2.internal.InternalPartitionedState;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalAggregatingState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/AbstractAggregatingState.java
similarity index 78%
rename from flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalAggregatingState.java
rename to flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/AbstractAggregatingState.java
index 3b1f85226a3..8ec735d18e0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalAggregatingState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/AbstractAggregatingState.java
@@ -23,6 +23,9 @@ import org.apache.flink.api.common.state.v2.AggregatingState;
import org.apache.flink.api.common.state.v2.StateFuture;
import org.apache.flink.runtime.asyncprocessing.StateRequestHandler;
import org.apache.flink.runtime.asyncprocessing.StateRequestType;
+import org.apache.flink.runtime.state.v2.internal.InternalAggregatingState;
+
+import java.util.Collection;
/**
* The default implementation of {@link AggregatingState}, which delegates all async requests to
@@ -33,18 +36,18 @@ import org.apache.flink.runtime.asyncprocessing.StateRequestType;
* @param <ACC> TThe type of the accumulator (intermediate aggregation state).
* @param <OUT> The type of the values that are returned from the state.
*/
-public class InternalAggregatingState<K, N, IN, ACC, OUT> extends InternalKeyedState<K, N, ACC>
- implements AggregatingState<IN, OUT> {
+public class AbstractAggregatingState<K, N, IN, ACC, OUT> extends AbstractKeyedState<K, N, ACC>
+ implements InternalAggregatingState<K, N, IN, ACC, OUT> {
protected final AggregateFunction<IN, ACC, OUT> aggregateFunction;
/**
- * Creates a new InternalKeyedState with the given asyncExecutionController and stateDescriptor.
+ * Creates a new AbstractKeyedState with the given asyncExecutionController and stateDescriptor.
*
* @param stateRequestHandler The async request handler for handling all requests.
* @param stateDescriptor The properties of the state.
*/
- public InternalAggregatingState(
+ public AbstractAggregatingState(
StateRequestHandler stateRequestHandler,
AggregatingStateDescriptor<IN, ACC, OUT> stateDescriptor) {
super(stateRequestHandler, stateDescriptor);
@@ -70,4 +73,14 @@ public class InternalAggregatingState<K, N, IN, ACC, OUT> extends InternalKeyedS
public void add(IN value) {
handleRequestSync(StateRequestType.AGGREGATING_ADD, value);
}
+
+ @Override
+ public StateFuture<Void> asyncMergeNamespaces(N target, Collection<N> sources) {
+ throw new UnsupportedOperationException("To be implemented.");
+ }
+
+ @Override
+ public void mergeNamespaces(N target, Collection<N> sources) {
+ throw new UnsupportedOperationException("To be implemented.");
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalKeyedState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/AbstractKeyedState.java
similarity index 88%
rename from flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalKeyedState.java
rename to flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/AbstractKeyedState.java
index ef2b0a18531..d1492410992 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalKeyedState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/AbstractKeyedState.java
@@ -24,10 +24,11 @@ import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.runtime.asyncprocessing.AsyncExecutionController;
import org.apache.flink.runtime.asyncprocessing.StateRequestHandler;
import org.apache.flink.runtime.asyncprocessing.StateRequestType;
+import org.apache.flink.runtime.state.v2.internal.InternalKeyedState;
/**
- * The {@code InternalKeyedState} is the root of the internal state type hierarchy, similar to the
- * {@link State} being the root of the public API state hierarchy.
+ * The {@code AbstractKeyedState} is the root of the abstract state implementation hierarchy,
+ * similar to the {@link State} being the root of the public API state hierarchy.
*
* <p>The public API state hierarchy is intended to be programmed against by Flink applications. The
* internal state hierarchy holds all the auxiliary methods that communicates with {@link
@@ -37,16 +38,16 @@ import org.apache.flink.runtime.asyncprocessing.StateRequestType;
* @param <V> The type of values kept internally in state.
*/
@Internal
-public abstract class InternalKeyedState<K, N, V> implements InternalPartitionedState<N> {
+public abstract class AbstractKeyedState<K, N, V> implements InternalKeyedState<K, N, V> {
protected final StateRequestHandler stateRequestHandler;
private final StateDescriptor<V> stateDescriptor;
/**
- * Creates a new InternalKeyedState with the given asyncExecutionController and stateDescriptor.
+ * Creates a new AbstractKeyedState with the given asyncExecutionController and stateDescriptor.
*/
- public InternalKeyedState(
+ public AbstractKeyedState(
StateRequestHandler stateRequestHandler, StateDescriptor<V> stateDescriptor) {
this.stateRequestHandler = stateRequestHandler;
this.stateDescriptor = stateDescriptor;
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalListState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/AbstractListState.java
similarity index 92%
rename from flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalListState.java
rename to flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/AbstractListState.java
index 92dc60751ea..8631f954237 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalListState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/AbstractListState.java
@@ -22,6 +22,7 @@ import org.apache.flink.api.common.state.v2.StateFuture;
import org.apache.flink.api.common.state.v2.StateIterator;
import org.apache.flink.runtime.asyncprocessing.StateRequestHandler;
import org.apache.flink.runtime.asyncprocessing.StateRequestType;
+import org.apache.flink.runtime.state.v2.internal.InternalListState;
import java.util.List;
@@ -32,10 +33,10 @@ import java.util.List;
* @param <K> The type of key the state is associated to.
* @param <V> The type of values kept internally in state.
*/
-public class InternalListState<K, N, V> extends InternalKeyedState<K, N, V>
- implements ListState<V> {
+public class AbstractListState<K, N, V> extends AbstractKeyedState<K, N, V>
+ implements InternalListState<K, N, V> {
- public InternalListState(
+ public AbstractListState(
StateRequestHandler stateRequestHandler, ListStateDescriptor<V> stateDescriptor) {
super(stateRequestHandler, stateDescriptor);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalMapState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/AbstractMapState.java
similarity index 95%
rename from flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalMapState.java
rename to flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/AbstractMapState.java
index 120b93e48d7..108b3128344 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalMapState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/AbstractMapState.java
@@ -23,6 +23,7 @@ import org.apache.flink.api.common.state.v2.StateIterator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.asyncprocessing.StateRequestHandler;
import org.apache.flink.runtime.asyncprocessing.StateRequestType;
+import org.apache.flink.runtime.state.v2.internal.InternalMapState;
import java.util.Iterator;
import java.util.Map;
@@ -35,10 +36,10 @@ import java.util.Map;
* @param <UK> The type of user key of this state.
* @param <V> The type of values kept internally in state.
*/
-public class InternalMapState<K, N, UK, V> extends InternalKeyedState<K, N, V>
- implements MapState<UK, V> {
+public class AbstractMapState<K, N, UK, V> extends AbstractKeyedState<K, N, V>
+ implements InternalMapState<K, N, UK, V> {
- public InternalMapState(
+ public AbstractMapState(
StateRequestHandler stateRequestHandler, MapStateDescriptor<UK, V> stateDescriptor) {
super(stateRequestHandler, stateDescriptor);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalReducingState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/AbstractReducingState.java
similarity index 96%
rename from flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalReducingState.java
rename to flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/AbstractReducingState.java
index 0c3b124a778..de7a19fd157 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalReducingState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/AbstractReducingState.java
@@ -23,6 +23,7 @@ import org.apache.flink.api.common.state.v2.StateFuture;
import org.apache.flink.core.state.StateFutureUtils;
import org.apache.flink.runtime.asyncprocessing.StateRequestHandler;
import org.apache.flink.runtime.asyncprocessing.StateRequestType;
+import org.apache.flink.runtime.state.v2.internal.InternalReducingState;
import java.util.ArrayList;
import java.util.Collection;
@@ -36,12 +37,12 @@ import java.util.List;
* @param <K> The type of key the state is associated to.
* @param <V> The type of values kept internally in state.
*/
-public class InternalReducingState<K, N, V> extends InternalKeyedState<K, N, V>
- implements ReducingState<V>, InternalMergingState<N> {
+public class AbstractReducingState<K, N, V> extends AbstractKeyedState<K, N, V>
+ implements InternalReducingState<K, N, V> {
protected final ReduceFunction<V> reduceFunction;
- public InternalReducingState(
+ public AbstractReducingState(
StateRequestHandler stateRequestHandler, ReducingStateDescriptor<V> stateDescriptor) {
super(stateRequestHandler, stateDescriptor);
this.reduceFunction = stateDescriptor.getReduceFunction();
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalValueState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/AbstractValueState.java
similarity index 90%
rename from flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalValueState.java
rename to flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/AbstractValueState.java
index 9a1ddfe5e83..0affbf415f3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalValueState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/AbstractValueState.java
@@ -22,6 +22,7 @@ import org.apache.flink.api.common.state.v2.ValueState;
import org.apache.flink.runtime.asyncprocessing.AsyncExecutionController;
import org.apache.flink.runtime.asyncprocessing.StateRequestHandler;
import org.apache.flink.runtime.asyncprocessing.StateRequestType;
+import org.apache.flink.runtime.state.v2.internal.InternalValueState;
/**
* A default implementation of {@link ValueState} which delegates all async requests to {@link
@@ -30,10 +31,10 @@ import org.apache.flink.runtime.asyncprocessing.StateRequestType;
* @param <K> The type of key the state is associated to.
* @param <V> The type of values kept internally in state.
*/
-public class InternalValueState<K, N, V> extends InternalKeyedState<K, N, V>
- implements ValueState<V> {
+public class AbstractValueState<K, N, V> extends AbstractKeyedState<K, N, V>
+ implements InternalValueState<K, N, V> {
- public InternalValueState(
+ public AbstractValueState(
StateRequestHandler stateRequestHandler, ValueStateDescriptor<V> valueStateDescriptor) {
super(stateRequestHandler, valueStateDescriptor);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalMergingState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalAggregatingState.java
similarity index 51%
copy from flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalMergingState.java
copy to flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalAggregatingState.java
index 3f5713bdb58..8b5d7b2dfa2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalMergingState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalAggregatingState.java
@@ -15,27 +15,18 @@
* limitations under the License.
*/
-package org.apache.flink.runtime.state.v2;
+package org.apache.flink.runtime.state.v2.internal;
-import org.apache.flink.api.common.state.v2.StateFuture;
-
-import java.util.Collection;
+import org.apache.flink.api.common.state.v2.AggregatingState;
/**
- * This class defines the internal interface for merging state.
+ * This class defines the internal interface for aggregating state.
*
- * @param <N> The type of the namespace
+ * @param <K> The type of key the state is associated to.
+ * @param <N> The namespace type.
+ * @param <IN> The type of the values that are added into the state.
+ * @param <ACC> The type of the accumulator (intermediate aggregation state).
+ * @param <OUT> The type of the values that are returned from the state.
*/
-public interface InternalMergingState<N> extends InternalPartitionedState<N> {
-
- /**
- * Merges the state of the current key for the given source namespaces into the state of the
- * target namespace.
- *
- * @param target The target namespace where the merged state should be stored.
- * @param sources The source namespaces whose state should be merged.
- */
- StateFuture<Void> asyncMergeNamespaces(N target, Collection<N> sources);
-
- void mergeNamespaces(N target, Collection<N> sources);
-}
+public interface InternalAggregatingState<K, N, IN, ACC, OUT>
+ extends InternalMergingState<K, N, IN, ACC, OUT, OUT>, AggregatingState<IN, OUT> {}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalMergingState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalAppendingState.java
similarity index 51%
copy from flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalMergingState.java
copy to flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalAppendingState.java
index 3f5713bdb58..6b9b323d8fe 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalMergingState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalAppendingState.java
@@ -15,27 +15,19 @@
* limitations under the License.
*/
-package org.apache.flink.runtime.state.v2;
+package org.apache.flink.runtime.state.v2.internal;
-import org.apache.flink.api.common.state.v2.StateFuture;
-
-import java.util.Collection;
+import org.apache.flink.api.common.state.v2.AppendingState;
/**
- * This class defines the internal interface for merging state.
+ * This class defines the internal interface for appending state.
*
- * @param <N> The type of the namespace
+ * @param <K> The type of key the state is associated to.
+ * @param <N> The namespace type.
+ * @param <IN> The type of the values that are added into the state.
+ * @param <SV> The type of the intermediate state.
+ * @param <OUT> The type of the values that are returned from the state.
+ * @param <SYNCOUT> Type of the value that can be retrieved from the state by synchronous interface.
*/
-public interface InternalMergingState<N> extends InternalPartitionedState<N> {
-
- /**
- * Merges the state of the current key for the given source namespaces into the state of the
- * target namespace.
- *
- * @param target The target namespace where the merged state should be stored.
- * @param sources The source namespaces whose state should be merged.
- */
- StateFuture<Void> asyncMergeNamespaces(N target, Collection<N> sources);
-
- void mergeNamespaces(N target, Collection<N> sources);
-}
+public interface InternalAppendingState<K, N, IN, SV, OUT, SYNCOUT>
+ extends InternalKeyedState<K, N, SV>, AppendingState<IN, OUT, SYNCOUT> {}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalPartitionedState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalKeyedState.java
similarity index 71%
copy from flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalPartitionedState.java
copy to flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalKeyedState.java
index 46e35b89838..e8f2e29008a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalPartitionedState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalKeyedState.java
@@ -16,19 +16,17 @@
* limitations under the License.
*/
-package org.apache.flink.runtime.state.v2;
+package org.apache.flink.runtime.state.v2.internal;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.state.v2.State;
-/** A state that is partitioned into different namespaces. */
+/**
+ * This class defines the internal interface for all keyed state.
+ *
+ * @param <K> The type of key the state is associated to.
+ * @param <N> The namespace type.
+ * @param <V> The type of the intermediate state.
+ */
@Internal
-public interface InternalPartitionedState<N> extends State {
-
- /**
- * Set current namespace and access state under specified namespace afterward.
- *
- * @param namespace the specified namespace
- */
- void setCurrentNamespace(N namespace);
-}
+public interface InternalKeyedState<K, N, V> extends InternalPartitionedState<N>, State {}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalPartitionedState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalListState.java
similarity index 67%
copy from flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalPartitionedState.java
copy to flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalListState.java
index 46e35b89838..efeccc7539a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalPartitionedState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalListState.java
@@ -16,19 +16,17 @@
* limitations under the License.
*/
-package org.apache.flink.runtime.state.v2;
+package org.apache.flink.runtime.state.v2.internal;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.api.common.state.v2.ListState;
-/** A state that is partitioned into different namespaces. */
+/**
+ * This class defines the internal interface for list state.
+ *
+ * @param <K> The type of key the state is associated to.
+ * @param <N> The namespace type.
+ * @param <V> The type of the intermediate state.
+ */
@Internal
-public interface InternalPartitionedState<N> extends State {
-
- /**
- * Set current namespace and access state under specified namespace afterward.
- *
- * @param namespace the specified namespace
- */
- void setCurrentNamespace(N namespace);
-}
+public interface InternalListState<K, N, V> extends InternalKeyedState<K, N, V>, ListState<V> {}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalPartitionedState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalMapState.java
similarity index 64%
copy from flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalPartitionedState.java
copy to flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalMapState.java
index 46e35b89838..e22d4836527 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalPartitionedState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalMapState.java
@@ -16,19 +16,19 @@
* limitations under the License.
*/
-package org.apache.flink.runtime.state.v2;
+package org.apache.flink.runtime.state.v2.internal;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.api.common.state.v2.MapState;
-/** A state that is partitioned into different namespaces. */
+/**
+ * This class defines the internal interface for map state.
+ *
+ * @param <K> The type of key the state is associated to.
+ * @param <N> The namespace type.
+ * @param <UK> Type of the keys in the state.
+ * @param <UV> Type of the values in the state.
+ */
@Internal
-public interface InternalPartitionedState<N> extends State {
-
- /**
- * Set current namespace and access state under specified namespace afterward.
- *
- * @param namespace the specified namespace
- */
- void setCurrentNamespace(N namespace);
-}
+public interface InternalMapState<K, N, UK, UV>
+ extends InternalKeyedState<K, N, UV>, MapState<UK, UV> {}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalMergingState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalMergingState.java
similarity index 59%
copy from flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalMergingState.java
copy to flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalMergingState.java
index 3f5713bdb58..d3ee2522b17 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalMergingState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalMergingState.java
@@ -15,8 +15,9 @@
* limitations under the License.
*/
-package org.apache.flink.runtime.state.v2;
+package org.apache.flink.runtime.state.v2.internal;
+import org.apache.flink.api.common.state.v2.MergingState;
import org.apache.flink.api.common.state.v2.StateFuture;
import java.util.Collection;
@@ -24,9 +25,15 @@ import java.util.Collection;
/**
* This class defines the internal interface for merging state.
*
- * @param <N> The type of the namespace
+ * @param <K> The type of key the state is associated to.
+ * @param <N> The namespace type.
+ * @param <IN> The type of the values that are added into the state.
+ * @param <SV> The type of the intermediate state.
+ * @param <OUT> The type of the values that are returned from the state.
+ * @param <SYNCOUT> Type of the value that can be retrieved from the state by synchronous interface.
*/
-public interface InternalMergingState<N> extends InternalPartitionedState<N> {
+public interface InternalMergingState<K, N, IN, SV, OUT, SYNCOUT>
+ extends InternalAppendingState<K, N, IN, SV, OUT, SYNCOUT>, MergingState<IN, OUT, SYNCOUT> {
/**
* Merges the state of the current key for the given source namespaces into the state of the
@@ -37,5 +44,12 @@ public interface InternalMergingState<N> extends InternalPartitionedState<N> {
*/
StateFuture<Void> asyncMergeNamespaces(N target, Collection<N> sources);
+ /**
+ * Merges the state of the current key for the given source namespaces into the state of the
+ * target namespace.
+ *
+ * @param target The target namespace where the merged state should be stored.
+ * @param sources The source namespaces whose state should be merged.
+ */
void mergeNamespaces(N target, Collection<N> sources);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalPartitionedState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalPartitionedState.java
similarity index 87%
copy from flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalPartitionedState.java
copy to flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalPartitionedState.java
index 46e35b89838..201699e5c4f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalPartitionedState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalPartitionedState.java
@@ -16,14 +16,13 @@
* limitations under the License.
*/
-package org.apache.flink.runtime.state.v2;
+package org.apache.flink.runtime.state.v2.internal;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.state.v2.State;
/** A state that is partitioned into different namespaces. */
@Internal
-public interface InternalPartitionedState<N> extends State {
+public interface InternalPartitionedState<N> {
/**
* Set current namespace and access state under specified namespace afterward.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalMergingState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalReducingState.java
similarity index 51%
rename from flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalMergingState.java
rename to flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalReducingState.java
index 3f5713bdb58..bdebbfc2b2d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalMergingState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalReducingState.java
@@ -15,27 +15,16 @@
* limitations under the License.
*/
-package org.apache.flink.runtime.state.v2;
+package org.apache.flink.runtime.state.v2.internal;
-import org.apache.flink.api.common.state.v2.StateFuture;
-
-import java.util.Collection;
+import org.apache.flink.api.common.state.v2.ReducingState;
/**
- * This class defines the internal interface for merging state.
+ * This class defines the internal interface for reducing state.
*
- * @param <N> The type of the namespace
+ * @param <K> The type of key the state is associated to.
+ * @param <N> The namespace type.
+ * @param <T> Type of the value in the operator state.
*/
-public interface InternalMergingState<N> extends InternalPartitionedState<N> {
-
- /**
- * Merges the state of the current key for the given source namespaces into the state of the
- * target namespace.
- *
- * @param target The target namespace where the merged state should be stored.
- * @param sources The source namespaces whose state should be merged.
- */
- StateFuture<Void> asyncMergeNamespaces(N target, Collection<N> sources);
-
- void mergeNamespaces(N target, Collection<N> sources);
-}
+public interface InternalReducingState<K, N, T>
+ extends InternalAggregatingState<K, N, T, T, T>, ReducingState<T> {}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalPartitionedState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalValueState.java
similarity index 66%
rename from flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalPartitionedState.java
rename to flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalValueState.java
index 46e35b89838..d87396848f8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalPartitionedState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalValueState.java
@@ -16,19 +16,17 @@
* limitations under the License.
*/
-package org.apache.flink.runtime.state.v2;
+package org.apache.flink.runtime.state.v2.internal;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.api.common.state.v2.ValueState;
-/** A state that is partitioned into different namespaces. */
+/**
+ * This class defines the internal interface for value state.
+ *
+ * @param <K> The type of key the state is associated to.
+ * @param <N> The namespace type.
+ * @param <V> Type of the value in the operator state.
+ */
@Internal
-public interface InternalPartitionedState<N> extends State {
-
- /**
- * Set current namespace and access state under specified namespace afterward.
- *
- * @param namespace the specified namespace
- */
- void setCurrentNamespace(N namespace);
-}
+public interface InternalValueState<K, N, V> extends InternalKeyedState<K, N, V>, ValueState<V> {}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionControllerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionControllerTest.java
index 8698e8d2843..b5f722e63e0 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionControllerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionControllerTest.java
@@ -35,7 +35,7 @@ import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.StateBackendTestUtils;
import org.apache.flink.runtime.state.VoidNamespace;
import org.apache.flink.runtime.state.VoidNamespaceSerializer;
-import org.apache.flink.runtime.state.v2.InternalValueState;
+import org.apache.flink.runtime.state.v2.AbstractValueState;
import org.apache.flink.runtime.state.v2.ValueStateDescriptor;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.Preconditions;
@@ -859,7 +859,7 @@ class AsyncExecutionControllerTest {
}
}
- static class TestValueState extends InternalValueState<String, String, Integer> {
+ static class TestValueState extends AbstractValueState<String, String, Integer> {
private final TestUnderlyingState underlyingState;
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/InternalAggregatingStateTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/AbstractAggregatingStateTest.java
similarity index 91%
rename from flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/InternalAggregatingStateTest.java
rename to flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/AbstractAggregatingStateTest.java
index da671c78a83..126638d3ee0 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/InternalAggregatingStateTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/AbstractAggregatingStateTest.java
@@ -24,8 +24,8 @@ import org.apache.flink.runtime.asyncprocessing.StateRequestType;
import org.junit.jupiter.api.Test;
-/** Tests for {@link InternalAggregatingState}. */
-class InternalAggregatingStateTest extends InternalKeyedStateTestBase {
+/** Tests for {@link AbstractAggregatingState}. */
+class AbstractAggregatingStateTest extends AbstractKeyedStateTestBase {
@Test
@SuppressWarnings({"unchecked"})
@@ -55,8 +55,8 @@ class InternalAggregatingStateTest extends InternalKeyedStateTestBase {
AggregatingStateDescriptor<Integer, Integer, Integer> descriptor =
new AggregatingStateDescriptor<>(
"testAggState", aggregator, BasicTypeInfo.INT_TYPE_INFO);
- InternalAggregatingState<String, Void, Integer, Integer, Integer> state =
- new InternalAggregatingState<>(aec, descriptor);
+ AbstractAggregatingState<String, Void, Integer, Integer, Integer> state =
+ new AbstractAggregatingState<>(aec, descriptor);
aec.setCurrentContext(aec.buildContext("test", "test"));
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/InternalKeyedStateTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/AbstractKeyedStateTestBase.java
similarity index 98%
rename from flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/InternalKeyedStateTestBase.java
rename to flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/AbstractKeyedStateTestBase.java
index f089afc497d..a6a62991da5 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/InternalKeyedStateTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/AbstractKeyedStateTestBase.java
@@ -52,8 +52,8 @@ import java.util.concurrent.atomic.AtomicReference;
import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
-/** Tests for the common/shared functionality of {@link InternalKeyedState}. */
-public class InternalKeyedStateTestBase {
+/** Tests for the common/shared functionality of {@link AbstractKeyedState}. */
+public class AbstractKeyedStateTestBase {
@SuppressWarnings({"rawtypes"})
AsyncExecutionController aec;
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/InternalListStateTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/AbstractListStateTest.java
similarity index 92%
rename from flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/InternalListStateTest.java
rename to flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/AbstractListStateTest.java
index 7f6679d2399..8f87102d50a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/InternalListStateTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/AbstractListStateTest.java
@@ -26,16 +26,16 @@ import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.List;
-/** Tests for {@link InternalListState}. */
-public class InternalListStateTest extends InternalKeyedStateTestBase {
+/** Tests for {@link AbstractListState}. */
+public class AbstractListStateTest extends AbstractKeyedStateTestBase {
@Test
@SuppressWarnings({"unchecked", "rawtypes"})
public void testEachOperation() {
ListStateDescriptor<Integer> descriptor =
new ListStateDescriptor<>("testState", BasicTypeInfo.INT_TYPE_INFO);
- InternalListState<String, Void, Integer> listState =
- new InternalListState<>(aec, descriptor);
+ AbstractListState<String, Void, Integer> listState =
+ new AbstractListState<>(aec, descriptor);
aec.setCurrentContext(aec.buildContext("test", "test"));
listState.asyncClear();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/InternalMapStateTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/AbstractMapStateTest.java
similarity index 94%
rename from flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/InternalMapStateTest.java
rename to flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/AbstractMapStateTest.java
index 52dd9b1c3ef..57eb4029664 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/InternalMapStateTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/AbstractMapStateTest.java
@@ -27,8 +27,8 @@ import org.junit.jupiter.api.Test;
import java.util.HashMap;
import java.util.Map;
-/** Tests for {@link InternalMapState}. */
-public class InternalMapStateTest extends InternalKeyedStateTestBase {
+/** Tests for {@link AbstractMapState}. */
+public class AbstractMapStateTest extends AbstractKeyedStateTestBase {
@Test
@SuppressWarnings({"unchecked", "rawtypes"})
@@ -36,8 +36,8 @@ public class InternalMapStateTest extends InternalKeyedStateTestBase {
MapStateDescriptor<String, Integer> descriptor =
new MapStateDescriptor<>(
"testState", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
- InternalMapState<String, Void, String, Integer> mapState =
- new InternalMapState<>(aec, descriptor);
+ AbstractMapState<String, Void, String, Integer> mapState =
+ new AbstractMapState<>(aec, descriptor);
aec.setCurrentContext(aec.buildContext("test", "test"));
mapState.asyncClear();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/InternalReducingStateTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/AbstractReducingStateTest.java
similarity index 96%
rename from flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/InternalReducingStateTest.java
rename to flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/AbstractReducingStateTest.java
index f923dd52128..8ceb44d0899 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/InternalReducingStateTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/AbstractReducingStateTest.java
@@ -40,8 +40,8 @@ import java.util.concurrent.CompletableFuture;
import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
-/** Tests for {@link InternalReducingState}. */
-public class InternalReducingStateTest extends InternalKeyedStateTestBase {
+/** Tests for {@link AbstractReducingState}. */
+public class AbstractReducingStateTest extends AbstractKeyedStateTestBase {
@Test
@SuppressWarnings({"unchecked"})
@@ -49,8 +49,8 @@ public class InternalReducingStateTest extends InternalKeyedStateTestBase {
ReduceFunction<Integer> reducer = Integer::sum;
ReducingStateDescriptor<Integer> descriptor =
new ReducingStateDescriptor<>("testState", reducer, BasicTypeInfo.INT_TYPE_INFO);
- InternalReducingState<String, Void, Integer> reducingState =
- new InternalReducingState<>(aec, descriptor);
+ AbstractReducingState<String, Void, Integer> reducingState =
+ new AbstractReducingState<>(aec, descriptor);
aec.setCurrentContext(aec.buildContext("test", "test"));
reducingState.asyncClear();
@@ -88,8 +88,8 @@ public class InternalReducingStateTest extends InternalKeyedStateTestBase {
100,
10000,
1);
- InternalReducingState<String, String, Integer> reducingState =
- new InternalReducingState<>(aec, descriptor);
+ AbstractReducingState<String, String, Integer> reducingState =
+ new AbstractReducingState<>(aec, descriptor);
aec.setCurrentContext(aec.buildContext("test", "test"));
aec.setCurrentNamespaceForState(reducingState, "1");
reducingState.asyncAdd(1);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/InternalValueStateTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/AbstractValueStateTest.java
similarity index 89%
rename from flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/InternalValueStateTest.java
rename to flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/AbstractValueStateTest.java
index 58278ecbd39..228ec3073ea 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/InternalValueStateTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/AbstractValueStateTest.java
@@ -23,16 +23,16 @@ import org.apache.flink.runtime.asyncprocessing.StateRequestType;
import org.junit.jupiter.api.Test;
-/** Tests for {@link InternalValueState}. */
-public class InternalValueStateTest extends InternalKeyedStateTestBase {
+/** Tests for {@link AbstractValueState}. */
+public class AbstractValueStateTest extends AbstractKeyedStateTestBase {
@Test
@SuppressWarnings({"unchecked", "rawtypes"})
public void testEachOperation() {
ValueStateDescriptor<Integer> descriptor =
new ValueStateDescriptor<>("testState", BasicTypeInfo.INT_TYPE_INFO);
- InternalValueState<String, Void, Integer> valueState =
- new InternalValueState<>(aec, descriptor);
+ AbstractValueState<String, Void, Integer> valueState =
+ new AbstractValueState<>(aec, descriptor);
aec.setCurrentContext(aec.buildContext("test", "test"));
valueState.asyncClear();
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStListState.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStListState.java
index 986dbb11ddc..7b5da10e960 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStListState.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStListState.java
@@ -29,7 +29,7 @@ import org.apache.flink.runtime.asyncprocessing.StateRequest;
import org.apache.flink.runtime.asyncprocessing.StateRequestHandler;
import org.apache.flink.runtime.asyncprocessing.StateRequestType;
import org.apache.flink.runtime.state.SerializedCompositeKeyBuilder;
-import org.apache.flink.runtime.state.v2.InternalListState;
+import org.apache.flink.runtime.state.v2.AbstractListState;
import org.apache.flink.runtime.state.v2.ListStateDescriptor;
import org.apache.flink.util.Preconditions;
@@ -41,7 +41,7 @@ import java.util.List;
import java.util.function.Supplier;
/**
- * The {@link InternalListState} implement for ForStDB.
+ * The {@link AbstractListState} implement for ForStDB.
*
* <p>{@link ForStStateBackend} must ensure that we set the {@link org.rocksdb.StringAppendOperator}
* on the column family that we use for our state since we use the {@code merge()} call.
@@ -50,7 +50,7 @@ import java.util.function.Supplier;
* @param <N> The type of the namespace.
* @param <V> The type of the value.
*/
-public class ForStListState<K, N, V> extends InternalListState<K, N, V>
+public class ForStListState<K, N, V> extends AbstractListState<K, N, V>
implements ListState<V>, ForStInnerTable<K, N, List<V>> {
/** The column family which this internal value state belongs to. */
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStMapState.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStMapState.java
index fe23949febb..2a2595cdfd6 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStMapState.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStMapState.java
@@ -30,7 +30,7 @@ import org.apache.flink.runtime.asyncprocessing.StateRequest;
import org.apache.flink.runtime.asyncprocessing.StateRequestHandler;
import org.apache.flink.runtime.asyncprocessing.StateRequestType;
import org.apache.flink.runtime.state.SerializedCompositeKeyBuilder;
-import org.apache.flink.runtime.state.v2.InternalMapState;
+import org.apache.flink.runtime.state.v2.AbstractMapState;
import org.apache.flink.runtime.state.v2.MapStateDescriptor;
import org.apache.flink.runtime.state.v2.StateDescriptor;
import org.apache.flink.util.Preconditions;
@@ -43,14 +43,14 @@ import java.util.Map;
import java.util.function.Supplier;
/**
- * The {@link InternalMapState} implement for ForStDB.
+ * The {@link AbstractMapState} implement for ForStDB.
*
* @param <K> The type of the key.
* @param <N> The type of the namespace.
* @param <UK> The type of the user key.
* @param <UV> The type of the user value.
*/
-public class ForStMapState<K, N, UK, UV> extends InternalMapState<K, N, UK, UV>
+public class ForStMapState<K, N, UK, UV> extends AbstractMapState<K, N, UK, UV>
implements MapState<UK, UV>, ForStInnerTable<K, N, UV> {
/** The column family which this internal value state belongs to. */
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStReducingState.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStReducingState.java
index 3fe2329bd7b..9861a2b451c 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStReducingState.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStReducingState.java
@@ -27,7 +27,7 @@ import org.apache.flink.runtime.asyncprocessing.StateRequest;
import org.apache.flink.runtime.asyncprocessing.StateRequestHandler;
import org.apache.flink.runtime.asyncprocessing.StateRequestType;
import org.apache.flink.runtime.state.SerializedCompositeKeyBuilder;
-import org.apache.flink.runtime.state.v2.InternalReducingState;
+import org.apache.flink.runtime.state.v2.AbstractReducingState;
import org.apache.flink.runtime.state.v2.ReducingStateDescriptor;
import org.apache.flink.util.Preconditions;
@@ -37,13 +37,13 @@ import java.io.IOException;
import java.util.function.Supplier;
/**
- * The {@link InternalReducingState} implement for ForStDB.
+ * The {@link AbstractReducingState} implement for ForStDB.
*
* @param <K> The type of the key.
* @param <N> The type of the namespace.
* @param <V> The type of the value.
*/
-public class ForStReducingState<K, N, V> extends InternalReducingState<K, N, V>
+public class ForStReducingState<K, N, V> extends AbstractReducingState<K, N, V>
implements ForStInnerTable<K, N, V> {
/** The column family which this internal value state belongs to. */
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStValueState.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStValueState.java
index 31d65f521a6..519f4634686 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStValueState.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStValueState.java
@@ -28,7 +28,7 @@ import org.apache.flink.runtime.asyncprocessing.StateRequest;
import org.apache.flink.runtime.asyncprocessing.StateRequestHandler;
import org.apache.flink.runtime.asyncprocessing.StateRequestType;
import org.apache.flink.runtime.state.SerializedCompositeKeyBuilder;
-import org.apache.flink.runtime.state.v2.InternalValueState;
+import org.apache.flink.runtime.state.v2.AbstractValueState;
import org.apache.flink.runtime.state.v2.ValueStateDescriptor;
import org.apache.flink.util.Preconditions;
@@ -38,12 +38,12 @@ import java.io.IOException;
import java.util.function.Supplier;
/**
- * The {@link InternalValueState} implement for ForStDB.
+ * The {@link AbstractValueState} implement for ForStDB.
*
* @param <K> The type of the key.
* @param <V> The type of the value.
*/
-public class ForStValueState<K, N, V> extends InternalValueState<K, N, V>
+public class ForStValueState<K, N, V> extends AbstractValueState<K, N, V>
implements ValueState<V>, ForStInnerTable<K, N, V> {
/** The column family which this internal value state belongs to. */
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStDBIterateOperationTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStDBIterateOperationTest.java
index 1eab3211224..1948ed5f129 100644
--- a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStDBIterateOperationTest.java
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStDBIterateOperationTest.java
@@ -25,7 +25,7 @@ import org.apache.flink.core.state.InternalStateFuture;
import org.apache.flink.runtime.asyncprocessing.StateRequestHandler;
import org.apache.flink.runtime.asyncprocessing.StateRequestType;
import org.apache.flink.runtime.state.VoidNamespace;
-import org.apache.flink.runtime.state.v2.InternalPartitionedState;
+import org.apache.flink.runtime.state.v2.internal.InternalPartitionedState;
import org.junit.jupiter.api.Test;
import org.rocksdb.RocksDB;
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStDBOperationTestBase.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStDBOperationTestBase.java
index df8c9ff42c3..14645a412e3 100644
--- a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStDBOperationTestBase.java
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStDBOperationTestBase.java
@@ -35,10 +35,10 @@ import org.apache.flink.runtime.state.KeyGroupRangeAssignment;
import org.apache.flink.runtime.state.SerializedCompositeKeyBuilder;
import org.apache.flink.runtime.state.VoidNamespace;
import org.apache.flink.runtime.state.VoidNamespaceSerializer;
-import org.apache.flink.runtime.state.v2.InternalPartitionedState;
import org.apache.flink.runtime.state.v2.ListStateDescriptor;
import org.apache.flink.runtime.state.v2.MapStateDescriptor;
import org.apache.flink.runtime.state.v2.ValueStateDescriptor;
+import org.apache.flink.runtime.state.v2.internal.InternalPartitionedState;
import org.apache.flink.util.function.BiFunctionWithException;
import org.apache.flink.util.function.FunctionWithException;
import org.apache.flink.util.function.ThrowingConsumer;
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateExecutorTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateExecutorTest.java
index 96f0c30a04f..00248ed4c7f 100644
--- a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateExecutorTest.java
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateExecutorTest.java
@@ -27,7 +27,7 @@ import org.apache.flink.runtime.asyncprocessing.StateRequestContainer;
import org.apache.flink.runtime.asyncprocessing.StateRequestType;
import org.apache.flink.runtime.state.KeyGroupRangeAssignment;
import org.apache.flink.runtime.state.VoidNamespace;
-import org.apache.flink.runtime.state.v2.InternalKeyedState;
+import org.apache.flink.runtime.state.v2.AbstractKeyedState;
import org.junit.jupiter.api.Test;
import org.rocksdb.WriteOptions;
@@ -230,7 +230,7 @@ class ForStStateExecutorTest extends ForStDBOperationTestBase {
@SuppressWarnings({"rawtypes", "unchecked"})
private <K, N, V, R> StateRequest<?, ?, ?, ?> buildStateRequest(
- InternalKeyedState<K, N, V> innerTable,
+ AbstractKeyedState<K, N, V> innerTable,
StateRequestType requestType,
K key,
V value,