WindowStore in ValueTransformer

Kafka Streams
1

Hello frens,
I want to deduplicate my records simillar to confluent tutorial:

but when I try to cast:

(WindowStore<String, Long>) processorContext.getStateStore(storeName);

error occurs:

Caused by: java.lang.ClassCastException: class org.apache.kafka.streams.processor.internals.AbstractReadWriteDecorator$KeyValueStoreReadWriteDecorator cannot be cast to class org.apache.kafka.streams.state.WindowStore (org.apache.kafka.streams.processor.internals.AbstractReadWriteDecorator$KeyValueStoreReadWriteDecorator and org.apache.kafka.streams.state.WindowStore are in unnamed module of loader 'app')
	at com.woblak.cstream.ingester.core.io.kafka.streams.job.cryptowattrade.transformer.CryptoTradeDistinctTransformer.init(CryptoTradeDistinctTransformer.java:26)
	at org.apache.kafka.streams.kstream.internals.AbstractStream$1$1.init(AbstractStream.java:118)
	at org.apache.kafka.streams.kstream.internals.KStreamTransformValues$KStreamTransformValuesProcessor.init(KStreamTransformValues.java:58)
	at org.apache.kafka.streams.processor.internals.ProcessorNode.lambda$init$0(ProcessorNode.java:97)
	at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:836)
	at org.apache.kafka.streams.processor.internals.ProcessorNode.init(ProcessorNode.java:94)

When I use KeyValueStore instead of WindowStore then everythin is fine.

hqdefault

2

Seems the added store is a KV-store, not a windowed store. Thus, you need to change the code that adds the state store (switch from adding KV-store, to add a windowed store).

The tutorial you mention, does add windowed store:

final StoreBuilder<WindowStore<String, Long>> dedupStoreBuilder = Stores.windowStoreBuilder(
                Stores.persistentWindowStore(storeName,
                        retentionPeriod,
                        windowSize,
                        false
                ),
                Serdes.String(),
                Serdes.Long());

        builder.addStateStore(dedupStoreBuilder);