KAFKA-6049: extend Kafka Streams Scala API for cogroup (KIP-150) #7847

mjsax · 2019-12-18T02:17:29Z

More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.

Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.

Committer Checklist (excluded from commit message)

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

mjsax · 2019-12-18T02:17:52Z

Call for review @bbejeck @vvcephei @wcarlson5

mjsax · 2019-12-18T02:19:58Z

streams/src/main/java/org/apache/kafka/streams/kstream/CogroupedKStream.java

-     * @return an instance of {@link SessionWindowedCogroupedKStream}
-     */
-    SessionWindowedCogroupedKStream<K, VOut> windowedBy(final SessionWindows sessionWindows);
-


Just re-ordered this method to have the same ordering over multiple classes (time-window always before session-window)

mjsax · 2019-12-18T02:21:16Z

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

  implicit def wrapSessionWindowedKStream[K, V](inner: SessionWindowedKStreamJ[K, V]): SessionWindowedKStream[K, V] =
    new SessionWindowedKStream[K, V](inner)

-  implicit def wrapTimeWindowedKStream[K, V](inner: TimeWindowedKStreamJ[K, V]): TimeWindowedKStream[K, V] =
-    new TimeWindowedKStream[K, V](inner)


Move this one up to get better overview/grouping. The three below are new

mjsax · 2019-12-18T02:21:53Z

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

@@ -525,7 +525,7 @@ class KStream[K, V](val inner: KStreamJ[K, V]) {
   * @see `org.apache.kafka.streams.kstream.KStream#join`
   * @deprecated since 2.4. Use [[KStream#join(KStream, ValueJoiner, JoinWindows, StreamJoined)]] instead.
   */
-  @deprecated
+  @deprecated("use join(KStream, ValueJoiner, JoinWindows, StreamJoined) instead", "2.4")


Just a side fix -- got a warning about it

mjsax · 2019-12-18T02:22:31Z

...ams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/SessionWindowedKStream.scala

@@ -21,15 +21,18 @@ package org.apache.kafka.streams.scala
 package kstream

 import org.apache.kafka.streams.kstream.internals.KTableImpl
-import org.apache.kafka.streams.kstream.{KTable => KTableJ, SessionWindowedKStream => SessionWindowedKStreamJ, _}


Only some side cleanup in this class

mjsax · 2019-12-18T02:24:02Z

.../src/main/scala/org/apache/kafka/streams/scala/kstream/SessionWindowedCogroupedKStream.scala

+  _
+}
+import org.apache.kafka.streams.scala.FunctionsCompatConversions._
+import org.apache.kafka.streams.scala.ImplicitConversions._


IntelliJ complains that this is an unused import, but without it, it does not compile at command line -- what might I do wrong?

You can explicitly import the implicit conversions that you're using. Actually, I would recommend not even using them implicitly, but just to explicitly invoke the relevant conversion methods when you need them. (i.e., implicit conversions are syntactic sugar for users of the public API, we should try to instead write explicit and maintainable code in our internal implementations)

It should be pretty obvious which ones you need if you just delete this line and see what fails to compile.

Based on this commend, I did a proper cleanup PR: #7852

We should merge the cleanup PR first, and I can rebase this PR afterwards.

mjsax · 2019-12-18T02:24:14Z

...ala/src/main/scala/org/apache/kafka/streams/scala/kstream/TimeWindowedCogroupedKStream.scala

+  _
+}
+import org.apache.kafka.streams.scala.FunctionsCompatConversions._
+import org.apache.kafka.streams.scala.ImplicitConversions._


Same question as above.

vvcephei

Hey @mjsax , thanks for adding this!

I answered your question. Otherwise, this looks reasonable, but I no longer test any Scala APIs that don't have "black box" tests... I.e., we need to write a test that uses the API just like users will to make sure it actually works the way we think it will.

vvcephei · 2019-12-18T16:39:51Z

.../src/main/scala/org/apache/kafka/streams/scala/kstream/SessionWindowedCogroupedKStream.scala

+  _
+}
+import org.apache.kafka.streams.scala.FunctionsCompatConversions._
+import org.apache.kafka.streams.scala.ImplicitConversions._


You can explicitly import the implicit conversions that you're using. Actually, I would recommend not even using them implicitly, but just to explicitly invoke the relevant conversion methods when you need them. (i.e., implicit conversions are syntactic sugar for users of the public API, we should try to instead write explicit and maintainable code in our internal implementations)

It should be pretty obvious which ones you need if you just delete this line and see what fails to compile.

bbejeck · 2019-12-19T16:17:24Z

@mjsax failures are relevant Execution failed for task ':spotlessScala'. scala formatting errors.

bbejeck

Thanks for the PR @mjsax. Overall this LGTM. I agree with @vvcephei about adding a new test(s) for the additions.

mjsax · 2019-12-21T01:27:48Z

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

 import org.scalatest.{FlatSpec, Matchers}
+import org.scalatestplus.junit.JUnitRunner


Some side cleanup. org.scalatest.junit.JUnitRunner is deprecated

mjsax · 2019-12-21T01:29:21Z

Updated this PR (rebased, added tests). Call for review @bbejeck @vvcephei

mjsax · 2020-01-06T21:24:52Z

@vvcephei @bbejeck Rebased to resolve merge conflict. Call for review.

bbejeck

Thanks for the updated PR @mjsax, LGTM

bbejeck · 2020-01-07T17:00:59Z

Java 8 failed with

kafka.admin.LeaderElectionCommandTest.testPathToJsonFile
kafka.api.PlaintextConsumerTest.testLowMaxFetchSizeForRequestAndPartition

Updated existing flaky test Jira tickets

Java 11 passed

retest this please

mjsax · 2020-01-07T20:51:11Z

Java 8 failed (one overlapping test failure):

kafka.admin.TopicCommandWithAdminClientTest.testCreateAlterTopicWithRackAware
kafka.api.PlaintextConsumerTest.testLowMaxFetchSizeForRequestAndPartition

Java 11 passed.

Retest this please.

mjsax · 2020-01-08T23:28:15Z

Java 11 passed.

Java 8 failed (different test):

org.apache.kafka.streams.integration.StandbyTaskCreationIntegrationTest.shouldCreateStandByTasksForMaterializedAndOptimizedSourceTables

Merging this.

mjsax added the streams label Dec 18, 2019

mjsax commented Dec 18, 2019

View reviewed changes

vvcephei reviewed Dec 18, 2019

View reviewed changes

bbejeck reviewed Dec 19, 2019

View reviewed changes

mjsax force-pushed the kafka-6049-cogroup-scala branch from c229ddb to 18176b9 Compare December 20, 2019 22:05

mjsax commented Dec 21, 2019

View reviewed changes

mjsax added 6 commits January 6, 2020 12:43

KAFKA-6049: extend Kafka Streams Scala API for cogroup (KIP-150)

01f4eea

Rebased

ef14644

add tests

96c5433

foo

43840a6

bar

9032c97

Rebased

7da4bdc

mjsax force-pushed the kafka-6049-cogroup-scala branch from 5c63bc9 to 7da4bdc Compare January 6, 2020 21:24

bbejeck approved these changes Jan 7, 2020

View reviewed changes

mjsax merged commit 1ccca5c into apache:trunk Jan 8, 2020

mjsax deleted the kafka-6049-cogroup-scala branch February 11, 2020 22:56

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

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-6049: extend Kafka Streams Scala API for cogroup (KIP-150) #7847

KAFKA-6049: extend Kafka Streams Scala API for cogroup (KIP-150) #7847

mjsax commented Dec 18, 2019

mjsax commented Dec 18, 2019

mjsax Dec 18, 2019

mjsax Dec 18, 2019

mjsax Dec 18, 2019

mjsax Dec 18, 2019

mjsax Dec 18, 2019

vvcephei Dec 18, 2019

mjsax Dec 19, 2019

mjsax Dec 18, 2019

vvcephei left a comment

vvcephei Dec 18, 2019

bbejeck commented Dec 19, 2019

bbejeck left a comment

mjsax Dec 21, 2019

mjsax commented Dec 21, 2019

mjsax commented Jan 6, 2020

bbejeck left a comment

bbejeck commented Jan 7, 2020

mjsax commented Jan 7, 2020

mjsax commented Jan 8, 2020

		import org.scalatest.{FlatSpec, Matchers}
		import org.scalatestplus.junit.JUnitRunner

KAFKA-6049: extend Kafka Streams Scala API for cogroup (KIP-150) #7847

KAFKA-6049: extend Kafka Streams Scala API for cogroup (KIP-150) #7847

Conversation

mjsax commented Dec 18, 2019

Committer Checklist (excluded from commit message)

mjsax commented Dec 18, 2019

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

vvcephei left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bbejeck commented Dec 19, 2019

bbejeck left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax commented Dec 21, 2019

mjsax commented Jan 6, 2020

bbejeck left a comment

Choose a reason for hiding this comment

bbejeck commented Jan 7, 2020

mjsax commented Jan 7, 2020

mjsax commented Jan 8, 2020