diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/AbstractRocksDBAppendingState.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/AbstractRocksDBAppendingState.java
index 8c0f4d7da7b22e323e31ccd3e5f870ec98c8be42..04e665a6e8797887a5c6e58f3ab9546ad23ac871 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/AbstractRocksDBAppendingState.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/AbstractRocksDBAppendingState.java
@@ -18,7 +18,6 @@
 
 package org.apache.flink.contrib.streaming.state;
 
-import org.apache.flink.api.common.state.State;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.runtime.state.internal.InternalAppendingState;
 import org.apache.flink.util.FlinkRuntimeException;
@@ -28,8 +27,8 @@ import org.rocksdb.RocksDBException;
 
 import java.io.IOException;
 
-abstract class AbstractRocksDBAppendingState <K, N, IN, SV, OUT, S extends State>
-	extends AbstractRocksDBState<K, N, SV, S>
+abstract class AbstractRocksDBAppendingState <K, N, IN, SV, OUT>
+	extends AbstractRocksDBState<K, N, SV>
 	implements InternalAppendingState<K, N, IN, SV, OUT> {
 
 	/**
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/AbstractRocksDBState.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/AbstractRocksDBState.java
index 2218bc0d3e0ee3fd6f07b6bae0e58222e067f158..b5ab996b1f67803b27b67601e76ac6451f501c1c 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/AbstractRocksDBState.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/AbstractRocksDBState.java
@@ -44,9 +44,8 @@ import java.io.IOException;
  * @param <K> The type of the key.
  * @param <N> The type of the namespace.
  * @param <V> The type of values kept internally in state.
- * @param <S> The type of {@link State}.
  */
-public abstract class AbstractRocksDBState<K, N, V, S extends State> implements InternalKvState<K, N, V>, State {
+public abstract class AbstractRocksDBState<K, N, V> implements InternalKvState<K, N, V>, State {
 
 	/** Serializer for the namespace. */
 	final TypeSerializer<N> namespaceSerializer;
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBAggregatingState.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBAggregatingState.java
index 2085fb86256a3d676cd9078b7a0be31ec7eee8cc..770c558c8278f7e127fdbfbd576193f8d45ac6d3 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBAggregatingState.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBAggregatingState.java
@@ -43,7 +43,7 @@ import java.util.Collection;
  * @param <R> The type of the value returned from the state
  */
 class RocksDBAggregatingState<K, N, T, ACC, R>
-	extends AbstractRocksDBAppendingState<K, N, T, ACC, R, AggregatingState<T, R>>
+	extends AbstractRocksDBAppendingState<K, N, T, ACC, R>
 	implements InternalAggregatingState<K, N, T, ACC, R> {
 
 	/** User-specified aggregation function. */
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBFoldingState.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBFoldingState.java
index 4d6635704b934c4fc8a4321efa727c3472de861a..c5e830fcedf0a69dc351c342d00b054aa74fe3e5 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBFoldingState.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBFoldingState.java
@@ -42,7 +42,7 @@ import org.rocksdb.ColumnFamilyHandle;
  */
 @Deprecated
 class RocksDBFoldingState<K, N, T, ACC>
-	extends AbstractRocksDBAppendingState<K, N, T, ACC, ACC, FoldingState<T, ACC>>
+	extends AbstractRocksDBAppendingState<K, N, T, ACC, ACC>
 	implements InternalFoldingState<K, N, T, ACC> {
 
 	/** User-specified fold function. */
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
index 700c5468c8c40be700db1ae909530230cfdd17bf..72445ca32dd4e1d83e35d76d98fcc7efc5b827ce 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
@@ -1400,7 +1400,7 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
 		}
 
 		@SuppressWarnings("unchecked")
-		AbstractRocksDBState<?, ?, SV, S> rocksDBState = (AbstractRocksDBState<?, ?, SV, S>) state;
+		AbstractRocksDBState<?, ?, SV> rocksDBState = (AbstractRocksDBState<?, ?, SV>) state;
 
 		Snapshot rocksDBSnapshot = db.getSnapshot();
 		try (
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java
index 6904c853fe310da965ad0fb08395133b15fb49d2..cedeed8753cb35065b80b8bf3136319807575e3f 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java
@@ -59,7 +59,7 @@ import static org.apache.flink.runtime.state.StateSnapshotTransformer.Collection
  * @param <V> The type of the values in the list state.
  */
 class RocksDBListState<K, N, V>
-	extends AbstractRocksDBState<K, N, List<V>, ListState<V>>
+	extends AbstractRocksDBState<K, N, List<V>>
 	implements InternalListState<K, N, V> {
 
 	/** Serializer for the values. */
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBMapState.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBMapState.java
index cb656b53b1b15296705eeb06e2bdc18a469502da..7abbe3d2c457e7358eed99120e5edcba49097ff9 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBMapState.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBMapState.java
@@ -64,7 +64,7 @@ import java.util.Map;
  * @param <UV> The type of the values in the map state.
  */
 class RocksDBMapState<K, N, UK, UV>
-	extends AbstractRocksDBState<K, N, Map<UK, UV>, MapState<UK, UV>>
+	extends AbstractRocksDBState<K, N, Map<UK, UV>>
 	implements InternalMapState<K, N, UK, UV> {
 
 	private static final Logger LOG = LoggerFactory.getLogger(RocksDBMapState.class);
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBReducingState.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBReducingState.java
index 138357b0d77c850b64cdd5376d41b3f135869e3f..69736ae03f702e3224196c4cc7a544351868fb5d 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBReducingState.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBReducingState.java
@@ -41,7 +41,7 @@ import java.util.Collection;
  * @param <V> The type of value that the state state stores.
  */
 class RocksDBReducingState<K, N, V>
-	extends AbstractRocksDBAppendingState<K, N, V, V, V, ReducingState<V>>
+	extends AbstractRocksDBAppendingState<K, N, V, V, V>
 	implements InternalReducingState<K, N, V> {
 
 	/** User-specified reduce function. */
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBValueState.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBValueState.java
index 0ca90d4a5212f67d71cdf1cc7753013c1be5e646..97b83dfa49c7a611a20735300c5d56eacbff192d 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBValueState.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBValueState.java
@@ -40,7 +40,7 @@ import java.io.IOException;
  * @param <V> The type of value that the state state stores.
  */
 class RocksDBValueState<K, N, V>
-	extends AbstractRocksDBState<K, N, V, ValueState<V>>
+	extends AbstractRocksDBState<K, N, V>
 	implements InternalValueState<K, N, V> {
 
 	/**