Make things more consistant and add Materialized helper with implicit Serdes

[joan38]

This helps on making the whole thing more typesafe since we use implicitly resolved Materialized.

Also this contains a change on the counts to use Materialized instead of the stateStore: String which is deprecated on the underlying Java API since for example it doesn't allow to configure caching.

Instead of mapping the values to convert them to scala.Long we can just cast the top type directly. This should save a tiny on performances.

[joan38]

Maybe we can just name this Materialized? I don't think it would conflict if we import org.apache.kafka.streams.kstream.{Materialized => JavaMaterialized

[joan38]

@debasishg I think mapValues without materialized is not persisting anything. Is that true? Or it's the same as the others?

[joan38]

Hi @debasishg, I know you're quite busy but do you know if you'll be able to have a look anytime soon?
I'm looking into contributing more but I don't want to open too many PRs at the same time.
Cheers

[debasishg]

I will surely look at them tomorrow. It's past midnight in my part of the world. Sorry for the delay.

[joan38]

Thanks @debasishg

[debasishg]

This will not work. count() that does not take a Materialized should not materialize to any store.

[debasishg]

I went through the changes. Implicit Materialized is something that won't work - we had a long discussion on this before. For APIs like count(), reduce() etc. there is a variant that does not materialize to stores. Also converting serdes to Materialized implicitly is not safe since Materialized can take lots of other parameters etc. caching, logging etc. which will have to be provided separately.

[joan38]

If I look at the API documentation of reduce() or count() without Materialized:

The result is written into a local KeyValueStore (which is basically an ever-updating materialized view) that can be queried using the provided queryableStoreName. Furthermore, updates to the store are sent downstream into a KTable changelog stream.

To me it's actually materializing to a store using the default serdes. And this will break at runtime if the default serdes are not the right ones.

The overall idea was that you have a default Materialized with only the serdes given implicitly so that we avoid using the default serdes and if you'd like (recommended) you can give one explicitly with more configuration (caching, logging...).

[debasishg]

The result is written into a local KeyValueStore

This result is the count value, which is Long and it uses a LongSerializer internally. The keySerde is used only if a specific store is requested to be built through a StoreSupplier. So we don't need a Materialized for def count() or def reduce().

[joan38]

I think you are right this would probably be useful for aggregate() but anyway this is much better to give a Materialized instead and this chamge would lead to an inconsistent API.

Instead we may provide a tiny helper for creating Materialized with implicit Serdes. See the new change I pushed. I'm not sure about the naming it might clash, let me know your thoughts.

[joan38]

This can now be closed as it's been opened on Kafka's repo with apache/kafka#4966