AyoubOm commented on code in PR #15790:
URL: https://github.com/apache/kafka/pull/15790#discussion_r1589953917
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java:
##########
@@ -173,21 +172,15 @@ protected Serde<V> prepareValueSerdeForStore(final
Serde<V> valueSerde, final Se
protected void initStoreSerde(final ProcessorContext context) {
final String storeName = name();
final String changelogTopic =
ProcessorContextUtils.changelogFor(context, storeName, Boolean.FALSE);
- serdes = new StateSerdes<>(
- changelogTopic,
- prepareKeySerde(keySerde, new SerdeGetter(context)),
- prepareValueSerdeForStore(valueSerde, new SerdeGetter(context))
- );
+ serdes = StoreSerdeInitializer.prepareStoreSerde(
+ context, storeName, changelogTopic, keySerde, valueSerde,
this::prepareValueSerdeForStore);
}
protected void initStoreSerde(final StateStoreContext context) {
final String storeName = name();
final String changelogTopic =
ProcessorContextUtils.changelogFor(context, storeName, Boolean.FALSE);
- serdes = new StateSerdes<>(
- changelogTopic,
- prepareKeySerde(keySerde, new SerdeGetter(context)),
- prepareValueSerdeForStore(valueSerde, new SerdeGetter(context))
- );
+ serdes = StoreSerdeInitializer.prepareStoreSerde(
+ context, storeName, changelogTopic, keySerde, valueSerde,
this::prepareValueSerdeForStore);
Review Comment:
I added a parameter function prepareValueSerde to be able to use the correct
function in children TimestampedStore classes. These don't directly use
`WrappingNullableUtils.prepareValueSerde` by overriding the behavior of
prepareValueSerdeForStore
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
