Implicit Serdes

[maver1ck]

Please add option to use implicit Serdes in this library,
Similar to:
https://github.com/ogirardot/typesafe-kafka-streams

[debasishg]

We are having thoughts on this as we speak. There was already a PR (#40) with implicit serdes.

We are in the process of thinking through the use cases as the handling of Serdes from the top level APIs has changed a lot in Kafka Streams 1.0. There is now an abstraction Serialized which is used for handling serialization of key/values. And the APIs have 2 types of signatures, one that takes a Serialized and uses them. And the other that uses the default serializers. We are thinking of how to unify them with implicits without losing the ability to use the default serializers.

[maver1ck]

By default serializers you mean those from default.key.serde and default.value.serde options ?
The question is if we need them at all? Probably not.

[debasishg]

If u look at the APIs KStream#groupByKey() and KStream#groupByKey(Serializable), the first one uses the default serializers while the second one uses the ones explicitly supplied. If we have implicit serializers and force the users to use them, then we lose out the first capability. Every time the user has to supply the serializers as implicit values. This is the topic we debated in the PR which I referenced earlier.

One option that we are thinking is to check the availability of the implicit values of the serializers and act accordingly .. if both serdes are available for key and value we use the second API, else use the first one.

WDYT ?

[maver1ck]

In the last question I just wanted to check if we're on the same page.

I'm thinking if using default serializers isn't equal to have them as implicit values ?
(even better because if serializer type is wrong we'll get an error during compilation not in the runtime)

So current setting:

config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

we'll change to:

implicit val stringSerde = Serdes.String()

[debasishg]

Indeed with implicits we have more type safety. The point is does this take away some feature from the underlying library ? We debated this as well in the PR.

So you think it's ok to take away this feature of setting the defaults as part of config and handle with implicits ?

[maver1ck]

But having function with implicit gives possibility to set explicit values on the runtime.
The only difference will be that we'll be using Serdes instead of Produced.with etc.

[debasishg]

Yes, we can set explicit values with Serdes. We can replace all APIs that take Serialized with implicit Serdes. But possibly we may have to keep the ones that take Produced or Consumed (in addition to the new ones that take implicit Serdes) as these 2 allow setting of other values besides the serdes.

[maver1ck]

Sounds good.
So for groupBy we will have two definitions:

def groupBy(function)(implicit keySerde: K, valueSerde: V)
def groupBy(function, Serialized.with(keySerde, valueSerde))

[debasishg]

We can debate if we need the second variant for Serialized if we have the first. But I think we need both the variants for functions that take Produced or Consumed as you can set other values besides the serdes with them.

[dhoepelman]

Maybe providing an opt-in that enables the default serialized behavior is a solution?

// This could also be Serde instead of Serialized
// The principle is the same but the implementation slightly more difficult
implicit case object ConfigSerialized extends Serialized[Nothing, Nothing] {
   //  throw UnsupportedOperationException in methods
}

def through(topic: String)(implicit serialized: Serialized[K,V]): KStreamS[K, V] =
  if(serialized eq ConfigSerialized)
    inner.through(topic)
  else
    inner.through(topic, serialized)

This way the default is the type-safe Scala-style implicit API, but if the users adds import streams.ConfigSerialized the default serializers can be used.

If you want the behavior as default you could make the signature

def through(topic: String)(implicit serialized: Serialized[K,V] = ConfigSerialized)

this won't need the import, but you don't get a compile-type error with a missing implicit if you forget to define it, which is not what Scala users expect.

[debasishg]

@dhoepelman - this was the idea I was toying with to provide both the options .. see my remark below in one of the earlier comments ..

One option that we are thinking is to check the availability of the implicit values of the serializers and act accordingly .. if both serdes are available for key and value we use the second API, else use the first one.

And you are correct that providing implicits on Serialized or Produced or Consumed calls for an easier implementation.

To everybody on this comment thread, how about the following strategy ..

1. we provide all implicits on Serialized or Produced or Consumed instead of Serde, because these are the preferred abstractions in Kafka Streams 1.0.0
2. use the strategy mentioned by @dhoepelman to handle default serializers. The default strategy will be using the implicits (recommended) but if someone wants to use the config serializers, she can do that with a little more effort of an additional import. This also follows the principle of making the recommended way easier to use.

[maver1ck]

Ad 1)
But then we need to provide O(n^2) implicit values instead of O(n), where n = number of types.

[dhoepelman]

Agree with 1 and 2

@maver1ck
The standard solution is to provide an implicit conversion (e.g. in play-json) if there is a good default option

implicit def SerializedFromSerde[K,V](implicit keySerde: Serde[K], valueSerde: Serde[V]): Serialized[K,V] = Serialized.`with`(keySerde, valueSerde)

[debasishg]

@dhoepelman This one may be tricky ..

implicit case object ConfigSerialized extends Serialized[Nothing, Nothing] {
   //  throw UnsupportedOperationException in methods
}

Serialized's constructor takes another Serialized and also the implicit match may not work because of (in)variance of Java type.

[debasishg]

I have started working on an implementation ..