[FLINK-11287] [rocksdb] RocksDBListState should be using registered s…

…erializer in state meta infos

This closes apache#7434.

flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java

@@ -19,7 +19,6 @@
 package org.apache.flink.contrib.streaming.state;
 
 import org.apache.flink.api.common.state.ListState;
-import org.apache.flink.api.common.state.ListStateDescriptor;
 import org.apache.flink.api.common.state.State;
 import org.apache.flink.api.common.state.StateDescriptor;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
@@ -77,19 +76,19 @@ class RocksDBListState<K, N, V>
  * @param namespaceSerializer The serializer for the namespace.
  * @param valueSerializer The serializer for the state.
  * @param defaultValue The default value for the state.
- * @param elementSerializer The serializer for elements of the list state.
  * @param backend The backend for which this state is bind to.
  */
  private RocksDBListState(
  ColumnFamilyHandle columnFamily,
  TypeSerializer<N> namespaceSerializer,
  TypeSerializer<List<V>> valueSerializer,
  List<V> defaultValue,
- TypeSerializer<V> elementSerializer,
  RocksDBKeyedStateBackend<K> backend) {
 
  super(columnFamily, namespaceSerializer, valueSerializer, defaultValue, backend);
- this.elementSerializer = elementSerializer;
+
+ ListSerializer<V> castedListSerializer = (ListSerializer<V>) valueSerializer;
+ this.elementSerializer = castedListSerializer.getElementSerializer();
  }
 
  @Override
@@ -281,7 +280,6 @@ static <E, K, N, SV, S extends State, IS extends S> IS create(
  registerResult.f1.getNamespaceSerializer(),
  (TypeSerializer<List<E>>) registerResult.f1.getStateSerializer(),
  (List<E>) stateDesc.getDefaultValue(),
- ((ListStateDescriptor<E>) stateDesc).getElementSerializer(),
  backend);
  }