Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/5691#discussion_r174423536
--- Diff:
flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapMapState.java
---
@@ -140,19 +159,28 @@ public boolean contains(UK userKey) {
}
@Override
- public byte[] getSerializedValue(K key, N namespace) throws IOException
{
- Preconditions.checkState(namespace != null, "No namespace
given.");
- Preconditions.checkState(key != null, "No key given.");
+ public byte[] getSerializedValue(
+ byte[] serializedKeyAndNamespace,
+ TypeSerializer<K> keySerializer,
+ TypeSerializer<N> namespaceSerializer,
+ TypeSerializer<HashMap<UK, UV>> valueSerializer) throws
Exception {
- HashMap<UK, UV> result = stateTable.get(key, namespace);
+ Preconditions.checkNotNull(serializedKeyAndNamespace,
"Serialized key and namespace");
- if (null == result) {
+ Tuple2<K, N> keyAndNamespace =
KvStateSerializer.deserializeKeyAndNamespace(
+ serializedKeyAndNamespace, keySerializer,
namespaceSerializer);
+
+ Map<UK, UV> result = stateTable.get(keyAndNamespace.f0,
keyAndNamespace.f1);
+
+ if (result == null) {
return null;
}
- TypeSerializer<UK> userKeySerializer =
stateDesc.getKeySerializer();
- TypeSerializer<UV> userValueSerializer =
stateDesc.getValueSerializer();
+ final HashMapSerializer<UK, UV> serializer =
(HashMapSerializer<UK, UV>) valueSerializer;
--- End diff --
The problem is that the Heap Backend uses the `HashMapSerializer` while
RocksDB uses `MapSerializer`. So here it should be sth like `TypeSerializer<?
extends Map<UK, UV>>` that then should somehow be casted to the appropriate
serializer somehow in order to get the key and the value serializer. The
solution that we have here is a bit cleaner, although the best would be to
remove the `HashMapSerializer` and have a `MapSerializer` in here.
---
