Count fix and Type alias refactor in Streams Scala API

[joan38]

Provide a tiny helper for creating Materialized with implicit Serdes.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[joan38]

@guozhangwang @debasishg Any thoughts on this?

[guozhangwang]

I think ByteArrayKeyValueStore is more succinct than KeyValueStore[Bytes, Array[Byte]]? cc @debasishg @seglo for a second thought.

[debasishg]

+1 @guozhangwang .. exactly for this reason the type synonym was introduced

[joan38]

I'm good with the type alias ByteArrayKeyValueStore but then we should do it consistently for the other types like SessionStore[Bytes, Array[Byte]]. I will then modify the PR to add the others.

[debasishg]

I would go defensive with implicits on Materialized. First, this abstraction doesn't add much value since it returns the Java abstraction. Also if someone doesn't wish to pass a serde to a Materialized, and use Materialized.as they can be inadvertently passed if implicits are in scope. This will lead to errors that are difficult to debug.

[seglo]

I don't believe there's much purpose to abstract Materialized to help with implicit SerDes now that the internal changes to propagate the SerDes for each of these aggregation operators has been completed in #4919. As @guozhangwang mentioned in that PR:

This is a meta explanation about the serde inheritance across multiple internal impl classes:

reduce: inherit the key and value serdes from the parent XXImpl class.
count: inherit the key serdes, enforce setting the Serdes.Long() for value serdes.
aggregate: inherit the key serdes, do not set for value serdes internally (line 92 here is for this case).

[joan38]

Indeed this doesn't bring too much value so I removed the add of the Materialized helper and left only some refactorings

[guozhangwang]

I'm just curious why we can save the casting of Long2long now?

[joan38]

Not sure why it's been done in the first place

[guozhangwang]

We need to double check if Scala 2.11 is compatible with java / scala Long (I'm not an expert in this case..)

If that is indeed the case, could the above count() overloaded function be updated as well?

[joan38]

There shouldn't be any difference between 2.12 and 2.11 here. Both are just doing boxing and unboxing in Scala long.

I'll write a test to but that on clear tomorrow.

[joan38]

Indeed this change is faulty. I reverted this and added a test to make sure we don't get in this in the future.

[debasishg]

Actually the other overload def count(): KTable[K, Long] doesn't compile without the mapValues. This tricked us into thinking that we need to do this for this overload as well. Possibly the type parameter Long in Materialized helps the compiler to infer the return type of Long and make the conversion. So we still need to do the conversion manually in the other overload of count.

[guozhangwang]

Other parts LGTM. Left two more minor comments.

[guozhangwang]

Should we revert the change here as well?

[joan38]

Indeed we should

[guozhangwang]

Ditto here.

[joan38]

This one would have triggered a StackOverflow looping on itself.

[joan38]

This one would have triggered a StackOverflow looping on itself.

[guozhangwang]

Thanks @joan38 ! Merged to trunk.

[joan38]

Thanks @guozhangwang