KAFKA-7396 Materialized, Serialized, Joined, Consumed and Produced with implicit Serdes #5551

joan38 · 2018-08-22T01:00:41Z

We want to make sure that we always have a serde for all Materialized, Serialized, Joined, Consumed and Produced.
For that we can make use of the implicit parameters in Scala.

KIP: https://cwiki.apache.org/confluence/display/KAFKA/KIP-365%3A+Materialized%2C+Serialized%2C+Joined%2C+Consumed+and+Produced+with+implicit+Serde

joan38 · 2018-08-22T09:08:36Z

This one is ready for review @guozhangwang @vvcephei @ijuma @debasishg

guozhangwang

LGTM. @vvcephei could you take another look?

debasishg · 2018-08-23T07:52:02Z

LGTM .. should we also do the same for Serialized ?

joan38 · 2018-08-23T07:58:34Z

@debasishg good point even tho I don't think we have to create manually any Serialized are we?
I guess if we do the Serialized, we should also do the Consumed and Produced?

joan38 · 2018-08-23T09:14:52Z

@mjsax Could you label this PR stream please?

mjsax · 2018-08-23T17:49:51Z

Wondering if this is a public API change? Do we need a KIP for this?

guozhangwang · 2018-08-23T19:28:04Z

I do not think it is a public API: users do not need to change any code they have if they do not want to, as this change is for allowing them to get some "syntax sugar".

joan38 · 2018-08-23T19:33:02Z

Ok I'm adding the Serialized, Consumed and Produced with implicit Serdes too

vvcephei

Hey @joan38 , It looks good!

I think the sugar classes might benefit from scaladoc explaining what they are, what they do, and providing a few examples.

vvcephei · 2018-08-23T17:46:19Z

streams/streams-scala/src/main/scala/org/apache/kafka/streams/package.scala


 package object scala {
  type ByteArrayKeyValueStore = KeyValueStore[Bytes, Array[Byte]]
  type ByteArraySessionStore = SessionStore[Bytes, Array[Byte]]
  type ByteArrayWindowStore = WindowStore[Bytes, Array[Byte]]
+
+  type Materialized[K, V, S <: StateStore] = org.apache.kafka.streams.kstream.Materialized[K, V, S]


Can you explain why this is necessary? Everything seems to compile and run fine when I remove it.

You must have not compiled tests.
You can't import the type from java and the object from Scala if they have the same name but different package, unless you give the fully qualified name.
The type here does the import for you in the right package.

vvcephei · 2018-08-23T21:14:04Z

@guozhangwang, I agree that this is not changing or breaking any public APIs, but it seems like it is adding new public APIs. We can think of it as similar to adding new convenience methods to the existing builders. As such, it does seem like it needs a KIP.

vvcephei · 2018-08-23T21:16:22Z

@joan38 Also, you might already have this in the works, but I think we should keep up the trend of having test coverage on all these additions.

We initially didn't have enough respect for how hard it is to just read a scala DSL and visually verify that it'll behave as expected. It might be over-learning, but I'd feel better if we could see the expected usage in some tests.

joan38 · 2018-08-24T02:02:02Z

As you expected I wrote the tests 😄 and ported the documentation.

joan38 · 2018-08-24T17:09:50Z

I forgot the Joined one

joan38 · 2018-08-25T18:13:31Z

The KIP has been created:
https://cwiki.apache.org/confluence/display/KAFKA/KIP-365%3A+Materialized%2C+Serialized%2C+Joined%2C+Consumed+and+Produced+with+implicit+Serde

tedyu · 2018-08-25T21:17:34Z

streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/ImplicitConversions.scala


  implicit def consumedFromSerde[K, V](implicit keySerde: Serde[K], valueSerde: Serde[V]): Consumed[K, V] =
-    Consumed.`with`(keySerde, valueSerde)
+    kstream.Consumed.`with`(keySerde, valueSerde)


What's the purpose for prefixing with kstream ?

If I remove the prefix, the code still compiles.

Thanks, this was an IDE refator that introduced it.

tedyu · 2018-08-25T21:17:58Z

streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/Consumed.scala

@@ -0,0 +1,83 @@
+/*
+ * Copyright (C) 2018 Joan Goyeau.


Please don't use individual's name.

Why is that?
I asked that earlier and it seemed ok: #5502 (comment)

Is there a special case for Alexis Seigneurin?:
https://github.com/apache/kafka/blob/trunk/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala#L3

Sorry I don't know much about those licensing.

The other code was release before it was merged into Apache Kafka. So the original copyright was with Alexis and Lightbend (ie, Lightbend copied Alexis code and later contributed it to AK).

For regular PRs, the contributor license agreement states, the ASF gets the copyright (from my understanding---also not a lawyer). Otherwise, it would also be a hassle to add copyright to any file that is changed... Also not a lawyer, but please read https://www.apache.org/licenses/icla.pdf (for committers it's required to file this -- for contributors consider the wiki information: https://cwiki.apache.org/confluence/display/KAFKA/Contributing+Code+Changes

When you contribute code, you affirm that the contribution is your original work and that you license the work to the project under the project's open source license. Whether or not you state this explicitly, by submitting any copyrighted material via pull request, email, or other means you agree to license the material under the project's open source license and warrant that you have the legal authority to do so.

tedyu · 2018-08-25T21:19:29Z

streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/Consumed.scala

+   *                           config will be used
+   * @param resetPolicy        the offset reset policy to be used. If `null` the default reset policy from config
+   *                           will be used
+   * @tparam K                 key type


It seems these two lines should be moved closer to line 31 where keySerde is documented

These are type param documentation. They are usually after the parameters.

From KStream:

* @tparam K Type of keys * @tparam V Type of values * @param inner The underlying Java abstraction for KStream

I guess that make sense to move them up since this is the order they are displayed in the signature. I will change that.

I'd recommend we adopt whatever are normal scaladoc conventions. I think I'm responsible for the KStream doc, but I wasn't aware of a convention.

tedyu · 2018-08-25T21:20:20Z

streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/Joined.scala

@@ -0,0 +1,44 @@
+/*
+ * Copyright (C) 2018 Joan Goyeau.


Drop author, please

See reply above #5551 (comment)

joan38 · 2018-08-28T22:14:42Z

@tedyu Yes I'm not sure if we should move the tests in the internal package just for the sake of making it accessible?
@bbejeck To me it sounds like there should be no XInternal and just immutable ones but I'm a Scala guy talking here.
I don't know much how we can do to restrict this but still be able to use it freely in tests.

vvcephei · 2018-08-29T18:35:25Z

One problem with Java is that the access control mechanisms are very coarse. Let's look at ProducedInternal. The reason it exists is to expose the keySerde(), valueSerde(), and streamPartitioner() getters. The Produced class intentionally doesn't expose the getters.

Regardless of whether Produced is immutable, if we add these getters to Produced, they would become part of the public API (which needs a KIP, etc).

It doesn't really matter whether ProducedInternal has a public constructor or not, since it's still in the internal package, which documents that it's "not a public interface".

It's better, of course, if we can avoid making the constructor public, since then the compiler backs up our package-based "this is not public" statement.

If this were scala, we could add private[streams] modifiers to these methods and just put them in Produced, but there's no way to do this in Java... c'est la vie.

joan38 · 2018-08-30T21:44:10Z

Do you know why this is not triggering Jenkins jobs?

vvcephei · 2018-08-31T14:28:28Z

I'm not sure...

Retest this, please.

joan38 · 2018-08-31T14:30:01Z

It worked 👍

vvcephei · 2018-08-31T14:38:46Z

Yeah, Jenkins must have fallen asleep. (or maybe there were no available workers or something)

vvcephei · 2018-08-31T14:39:49Z

It looks like the bounds may be backwards:

00:04:36.049 /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk8-scala2.11/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/Produced.scala:60: type mismatch;
00:04:36.049  found   : org.apache.kafka.streams.processor.StreamPartitioner[_$1,_$2] where type _$2 <: V, type _$1 <: K
00:04:36.049  required: org.apache.kafka.streams.processor.StreamPartitioner[_ >: K, _ >: V]
00:04:36.049 Note: _$1 <: Any, but Java-defined trait StreamPartitioner is invariant in type K.
00:04:36.049 You may wish to investigate a wildcard type such as `_ <: Any`. (SLS 3.2.10)
00:04:36.049 Note: _$2 <: Any, but Java-defined trait StreamPartitioner is invariant in type V.
00:04:36.049 You may wish to investigate a wildcard type such as `_ <: Any`. (SLS 3.2.10)
00:04:36.049     ProducedJ.`with`(keySerde, valueSerde, partitioner)

joan38 · 2018-08-31T14:44:44Z

Indeed, I will correct this.

joan38 · 2018-08-31T23:03:42Z

@vvcephei I just corrected it and now see:

Error:(60, 22) type mismatch;
 found   : org.apache.kafka.common.serialization.Serde[K]
 required: org.apache.kafka.common.serialization.Serde[_$1] where type _$1 >: K
Note: K <: _$1, but Java-defined trait Serde is invariant in type T.
You may wish to investigate a wildcard type such as `_ <: _$1`. (SLS 3.2.10)
    ProducedJ.`with`(keySerde, valueSerde, partitioner)

Again, I don't understand how can we have K for part or the function and _ >: K for the other part of the function. To me it should be all on the same type.

guozhangwang · 2018-09-01T18:38:32Z

streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/Consumed.scala

+  def `with`[K, V](
+    timestampExtractor: TimestampExtractor,
+    resetPolicy: Topology.AutoOffsetReset
+  )(implicit keySerde: Serde[K], valueSerde: Serde[V]): ConsumedJ[K, V] =


For my own education: why we only need implicit for the first parameter, instead of having it for both key and value serdes?

The implicit keyword applies to the whole set of parameters, not just to the first parameter.

@guozhangwang you are right that the declaration syntax of implicit parameters in Scala is confusing.
See this proposal scala/scala3#1260 for Scala 3:

The syntax (implicit x: T, y: U) is a bit strange in that implicit conceptually scopes over
x and y but looks like a modifier for just x.

That would make it clear that the implicit is applied to the whole group of curried parameters.

guozhangwang · 2018-09-01T18:39:46Z

streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/TopologyTest.scala

@@ -180,7 +191,7 @@ class TopologyTest extends JUnitSuite {

      // Compute the total per region by summing the individual click counts per region.
      val clicksPerRegion: KTableJ[String, JLong] = clicksByRegion
-        .groupByKey(Serialized.`with`(Serdes.String, Serdes.JavaLong))
+        .groupByKey(kstream.Serialized.`with`[String, JLong])


Why need the kstream. prefix here?

Wrong refactoring from my IDE. I will correct this.

joan38 · 2018-09-07T14:36:55Z

Are we good with this?

vvcephei · 2018-09-07T15:45:19Z

Hey @joan38 ,

You need to close the vote email thread and update the KIP status. I sent a message with the next steps to the vote thread for KIP-365.

vvcephei · 2018-09-07T16:32:42Z

streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/kstream/ConsumedTest.scala

+    val consumed: Consumed[String, Long] = Consumed.`with`[String, Long]
+
+    val internalConsumed = new ConsumedInternal(consumed)
+    internalConsumed.keySerde shouldBe Serdes.String


Gah. I just now realized that's it's not ok to be caching these serdes. It's not part of your change, though. I'll send a separate PR to fix it.

just FYI: https://issues.apache.org/jira/browse/KAFKA-7386 (I've assigned it to myself)

vvcephei

LGTM! Thanks!

We can merge as soon as the KIP is finalized. /cc @guozhangwang

mjsax · 2018-09-11T01:12:28Z

Btw: all KIPs should be backed by a Jira and the PR should have the ticket number in the title. Please update the PR and KIP accordingly after you created the Jira.

Also, this PR seems to need a rebase. Thx.

… Serdes

joan38 · 2018-09-11T11:19:45Z

@mjsax @vvcephei All done.
Should we go ahead and merge?

…ith implicit Serdes (apache#5551) We want to make sure that we always have a serde for all Materialized, Serialized, Joined, Consumed and Produced. For that we can make use of the implicit parameters in Scala. KIP: https://cwiki.apache.org/confluence/display/KAFKA/KIP-365%3A+Materialized%2C+Serialized%2C+Joined%2C+Consumed+and+Produced+with+implicit+Serde Reviewers: John Roesler <[email protected]>, Matthias J. Sax <[email protected]>, Bill Bejeck <[email protected]>, Guozhang Wang <[email protected]>, Ted Yu <[email protected]>

joan38 mentioned this pull request Aug 22, 2018

Serdes mandatory for Materialized #5541

Closed

guozhangwang reviewed Aug 22, 2018

View reviewed changes

guozhangwang added the streams label Aug 23, 2018

joan38 changed the title ~~Materialized with implicit Serdes~~ Materialized, Serialized, Consumed and Produced with implicit Serdes Aug 23, 2018

joan38 force-pushed the implicit-serde-mater branch 2 times, most recently from fe842a0 to e72c98d Compare August 23, 2018 20:54

vvcephei reviewed Aug 23, 2018

View reviewed changes

joan38 force-pushed the implicit-serde-mater branch from e72c98d to 97498e0 Compare August 24, 2018 02:00

joan38 force-pushed the implicit-serde-mater branch from 97498e0 to b07bea9 Compare August 24, 2018 08:16

joan38 changed the title ~~Materialized, Serialized, Consumed and Produced with implicit Serdes~~ Materialized, Serialized, Joined, Consumed and Produced with implicit Serdes Aug 24, 2018

joan38 force-pushed the implicit-serde-mater branch from b07bea9 to 529a02a Compare August 25, 2018 18:06

joan38 changed the title ~~Materialized, Serialized, Joined, Consumed and Produced with implicit Serdes~~ @joan38 Materialized, Serialized, Joined, Consumed and Produced with implicit Serdes Aug 25, 2018

joan38 changed the title ~~@joan38 Materialized, Serialized, Joined, Consumed and Produced with implicit Serdes~~ KIP-365 Materialized, Serialized, Joined, Consumed and Produced with implicit Serdes Aug 25, 2018

joan38 force-pushed the implicit-serde-mater branch from 529a02a to 32c2caa Compare August 25, 2018 18:46

tedyu reviewed Aug 25, 2018

View reviewed changes

joan38 force-pushed the implicit-serde-mater branch 3 times, most recently from 95606b1 to 6666a53 Compare August 26, 2018 02:12

joan38 force-pushed the implicit-serde-mater branch from b4140d5 to c7db46b Compare August 28, 2018 22:15

joan38 force-pushed the implicit-serde-mater branch from c7db46b to 8e4777f Compare August 29, 2018 22:55

joan38 force-pushed the implicit-serde-mater branch from 8e4777f to a91cb32 Compare August 30, 2018 21:51

joan38 force-pushed the implicit-serde-mater branch from a91cb32 to 3ee9b4f Compare August 31, 2018 23:00

guozhangwang reviewed Sep 1, 2018

View reviewed changes

joan38 force-pushed the implicit-serde-mater branch 2 times, most recently from 92f9228 to a019556 Compare September 4, 2018 23:03

vvcephei reviewed Sep 7, 2018

View reviewed changes

vvcephei approved these changes Sep 7, 2018

View reviewed changes

joan38 added 2 commits September 11, 2018 09:04

Materialized, Serialized, Joined, Consumed and Produced with implicit…

1843bc3

… Serdes

Fix compilation

5ba0eb5

joan38 force-pushed the implicit-serde-mater branch from a019556 to 5ba0eb5 Compare September 11, 2018 08:04

joan38 changed the title ~~KIP-365 Materialized, Serialized, Joined, Consumed and Produced with implicit Serdes~~ KAFKA-7396 Materialized, Serialized, Joined, Consumed and Produced with implicit Serdes Sep 11, 2018

guozhangwang merged commit acd3858 into apache:trunk Sep 11, 2018

joan38 deleted the implicit-serde-mater branch September 12, 2018 07:20

mjsax added the kip Requires or implements a KIP label Jun 12, 2020

KAFKA-7396 Materialized, Serialized, Joined, Consumed and Produced with implicit Serdes #5551

KAFKA-7396 Materialized, Serialized, Joined, Consumed and Produced with implicit Serdes #5551

Conversation

joan38 commented Aug 22, 2018 • edited Loading

joan38 commented Aug 22, 2018

guozhangwang left a comment

Choose a reason for hiding this comment

debasishg commented Aug 23, 2018

joan38 commented Aug 23, 2018 • edited Loading

joan38 commented Aug 23, 2018 • edited Loading

mjsax commented Aug 23, 2018

guozhangwang commented Aug 23, 2018

joan38 commented Aug 23, 2018

vvcephei left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vvcephei commented Aug 23, 2018

vvcephei commented Aug 23, 2018

joan38 commented Aug 24, 2018

joan38 commented Aug 24, 2018

joan38 commented Aug 25, 2018

Choose a reason for hiding this comment

joan38 Aug 26, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

joan38 Aug 26, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

joan38 commented Aug 28, 2018 • edited Loading

vvcephei commented Aug 29, 2018

joan38 commented Aug 30, 2018

vvcephei commented Aug 31, 2018

joan38 commented Aug 31, 2018

vvcephei commented Aug 31, 2018

vvcephei commented Aug 31, 2018

joan38 commented Aug 31, 2018

joan38 commented Aug 31, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

joan38 Sep 5, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

joan38 commented Sep 7, 2018

vvcephei commented Sep 7, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vvcephei left a comment

Choose a reason for hiding this comment

mjsax commented Sep 11, 2018

joan38 commented Sep 11, 2018

joan38 commented Aug 22, 2018 •

edited

Loading

joan38 commented Aug 23, 2018 •

edited

Loading

joan38 commented Aug 23, 2018 •

edited

Loading

joan38 Aug 26, 2018 •

edited

Loading

joan38 Aug 26, 2018 •

edited

Loading

joan38 commented Aug 28, 2018 •

edited

Loading

joan38 Sep 5, 2018 •

edited

Loading