Use default aggSerde if no user-overridden is provided in RocksDBSessionStore
RocksDBSessionStore wasn't properly using the default aggSerde if no Serde was supplied.
Author: Kyle Winkelman
Reviewers: Damian Guy, Guozhang Wang
Closes from KyleWinkelman/RocksDBSessionStore-fix-aggSerde-use
@@ -106,6 +106,6 @@ class RocksDBSessionStore<K, AGG> extends WrappedStateStore.AbstractStateStore i
@Override
public void put(final Windowed<K> sessionKey, final AGG aggregate) {
- bytesStore.put(SessionKeySerde.toBinary(sessionKey, serdes.keySerializer(), topic), aggSerde.serializer().serialize(topic, aggregate));
+ bytesStore.put(SessionKeySerde.toBinary(sessionKey, serdes.keySerializer(), topic), serdes.rawValue(aggregate));
}
}
