KAFKA-7027: Add an overload build method in scala #6373

massimosiani · 2019-03-05T11:06:02Z

The Java API can pass a Properties object to StreamsBuilder#build, to allow, e.g., topology optimization, while the Scala API does not yet. The latter only delegates the work to the underlying Java implementation.

Committer Checklist (excluded from commit message)

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

massimosiani · 2019-03-05T11:07:05Z

I didn' find a related issue, please let me know if I missed it (or missed a related PR), or if I should do anything else before submitting this.
Thank you!

bbejeck · 2019-03-07T23:17:25Z

@massimosiani thanks for the contribution! This actually is related to KIP-312 (Jira https://issues.apache.org/jira/browse/KAFKA-7027) so I've updated the title.

Also, can you rebase this to trunk? Then we can cherry-pick to 2.2

bbejeck

Thanks for the contribution @massimosiani!
This PR looks good to me.

We should add at least one test using the new builder overload.

bbejeck · 2019-03-07T23:28:12Z

\cc any of (@guozhangwang, @mjsax, @vvcephei, and @ableegoldman) for a second review

massimosiani · 2019-03-08T10:15:14Z

Hi @bbejeck, thanks for the review!
I have changed the base branch for the PR to trunk, and added a test case identical to the Simple one in TopologyTest.scala with the new API.
In the test, the TOPOLOGY_OPTIMIZATION property is passed, because that's the main goal of this PR.

Let me know if there is anything else I can do.
Thank you!

bbejeck

Thanks @massimosiani LGTM

bbejeck · 2019-03-08T15:17:24Z

Java 8 failed with

org.apache.kafka.common.network.SslSelectorTest.testCloseOldestConnectionWithMultipleStagedReceives
kafka.api.UserQuotaTest.testThrottledProducerConsumer
kafka.network.SocketServerTest.testConnectionRateLimit

Java 11 passed
retest this please

vvcephei

Hi @massimosiani , Thanks for the contribution!

The implementation looks "obviously correct", but I had some comments on the test. The biggest concern is that the equality comparison of TopologyDescription is currently broken.

As I said, the implementation looks right, but the test would have value as a regression test, to make sure we don't break the logic in the future.

Thanks again,
-John

vvcephei · 2019-03-08T15:54:17Z

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

+      val streamBuilder = new StreamsBuilderJ
+      val textLines = streamBuilder.stream[String, String](inputTopic)
+
+      val _: KStreamJ[String, String] = textLines.flatMapValues(


Just curious, what's the advantage of assigning to _ instead of just ignoring the result of the builder calls?

Hi, @vvcephei, there are other tests in that class, and I wanted to preserve the general structure. Probably, they should be clean up. Removed from my code.

vvcephei · 2019-03-08T15:54:23Z

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

+    }
+
+    // should match
+    assertEquals(getTopologyScala, getTopologyJava)


This comparison actually is invalid. No reason you should know that, though...

A contributor recently pointed it out (see #6210).

As a workaround, can we try comparing the result of TopologyDescription#toString instead?

With the new test, this command works:
gradle streams:streams-scala:test --tests TopologyTest
(gradle 5.2.1 on MacOS)

@massimosiani @vvcephei is correct about the test

This comparison actually is invalid. No reason you should know that, though...
A contributor recently pointed it out (see #6210).
As a workaround, can we try comparing the result of TopologyDescription#toString instead?

When comparing the getTopologyScala.toString and getTopologyJava.toString the test fails. But it's not your fault though, I'm sure you have just exposed a bug! Thanks and I've created a Jira for it (https://issues.apache.org/jira/browse/KAFKA-8101) and when that is fixed we can get this merged.

~~EDIT: This looks like a bug with the Streams Scala API in general as when using StreamsConfig.NO_OPTIMIZATION or not passing in Properties to builder.build() the test still fails.~~

vvcephei · 2019-03-08T16:01:44Z

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

+      val _: KStream[String, String] =
+        textLines.flatMapValues(v => pattern.split(v.toLowerCase))
+
+      streamBuilder.build(props).describe()


It seems like, if the intent is to make sure the optimization property actually gets applied, we'd need to use a topology that would change under optimization. Perhaps you can copy the topology (and verification logic) from org.apache.kafka.streams.integration.RepartitionOptimizingIntegrationTest?

Alternatively, it might be sufficient to make it a unit test by injecting a mocked Java builder and just verify that the Scala layer passes the properties down correctly... Not sure how straightforward that would be, though.

It's been a pain to write the "Java" version (that's probably because of me being new to kafka), but I've done it.

bbejeck · 2019-03-13T22:15:40Z

@massimosiani After looking over the Scala API the issue for the different topology between Java and Scala is here https://github.com/apache/kafka/blob/trunk/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KGroupedStream.scala#L51 .

KGroupedStream#count in Scala needs to add an additional mapValues proceessor to convert from a Java Long to a Scala Long, so a count operation won't have the same topology.

Thanks for your hard work so far and here's what I'd suggest going forward.

Change the example to have 3 streams and you call either selectKey or map on the first stream instance. Then perform two joins,

val sourceStream = builder.stream(inputTopic, Consumed.`with`(Serdes.String, Serdes.String))
val mappedStream: KStreamJ[String, String] =sourceStream.map((k: String, v: String) => ...))

val stream2 = ...
val stream3 = ...

val join1 = mappedStream.join(stream2....) 
val join2 - mappedStream.join(stream3,...)

This way with optimization turned on we should end up with only one repartition topic and since there is no aggregation operation the toplogies should end up the same.

We'll stil want the assertion step to be assertEquals(getTopologyScala.toString, getTopologyJava.toString)

massimosiani · 2019-03-14T10:23:52Z

Hi @bbejeck, thank you.
I am not sure whether it makes sense, but I also added an assertion to check that the topology without optimization is different from the optimized one, so I am actually testing an optimized topology.
I had to add the explicit NO_OPTIMIZE property so we are not checking the default behavior of build().
Let me know if that makes sense at all.
Thank you!

bbejeck

@massimosiani thanks for the updates.

I am not sure whether it makes sense, but I also added an assertion to check that the topology without optimization is different from the optimized one, so I am actually testing an optimized topology.

Yes, that makes sense, but I think to compare both optimized and unoptimized Scala/Java topologies has merit, I've left a comment regarding that.

Also, the build failure is related. The code you have here works for Scala version 2.12 but Scala 2.11 doesn't support Java 8 lamdas so the Java 8 Scala 2.11 build is failing.

You'll need to change initializer , aggregator etc. to have Java anonymous classes

val initializer: Initializer[Integer] =  new Initializer[Integer] {
     override def apply(): Integer = 0
 }
 val aggregator: Aggregator[String, String, Integer] = new Aggregator[String, String, Integer] {
     override def apply(key: String, value: String, aggregate: Integer): Integer =  aggregate + value.length
 }

This applies to all the operators on lines 345 to 353 I didn't list them out here
If you run ./gradlew -PscalaVersion=2.11 :streams:streams-scala:compileTestScala you'll see what all the errors are.

Thanks for your hard work and patience on this!

bbejeck · 2019-03-14T19:10:14Z

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

+      builder
+    }
+
+    assertNotEquals(getTopologyScala.build(props).describe.toString,


I like the idea here, but I'm thinking it's a good idea to compare unoptimized Scala topology to the unoptimized Java topology so I'd change this line to

assertEquals(getTopologyScala.build(propsNoOptimization).describe.toString, getTopologyScala.build(propsNoOptimization).describe.toString)

What about adding your assertion instead of substituting to line 399? Do you agree?

What about adding your assertion instead of substituting to line 399?

Yes, that works as well

bbejeck

Thanks for the updates @massimosiani. Changes LGTM pending Jenkins passing

bbejeck · 2019-03-14T23:44:45Z

Moving to new comment to make sure this gets picked up

@massimosiani one last thing I just noticed, do you mind changing the assertEquals on line 432-433
from:

assertEquals(getTopologyScala.build(propsNoOptimization).describe.toString, getTopologyScala.build(propsNoOptimization).describe.toString)
to:

assertEquals(getTopologyScala.build(propsNoOptimization).describe.toString, getTopologyJava.build(propsNoOptimization).describe.toString)

My apologies as this is my error from my previous comment.

massimosiani · 2019-03-15T08:16:04Z

Moving to new comment to make sure this gets picked up

@massimosiani one last thing I just noticed, do you mind changing the assertEquals on line 432-433
from:

assertEquals(getTopologyScala.build(propsNoOptimization).describe.toString, getTopologyScala.build(propsNoOptimization).describe.toString)
to:

assertEquals(getTopologyScala.build(propsNoOptimization).describe.toString, getTopologyJava.build(propsNoOptimization).describe.toString)

My apologies as this is my error from my previous comment.

My fault, I took it for granted and did not read it carefully.

vvcephei

LGTM! Thanks, @massimosiani !

bbejeck · 2019-03-15T14:57:21Z

Thanks for the contribution @massimosiani!

Merged #6373 to trunk

The Java API can pass a Properties object to StreamsBuilder#build, to allow, e.g., topology optimization, while the Scala API does not yet. The latter only delegates the work to the underlying Java implementation. Reviewers: John Roesler <john@confluent.io>, Bill Bejeck <bbejeck@gmail.com>

bbejeck · 2019-03-15T15:51:57Z

cherry-picked to 2.2, and 2.1

* apache/trunk: MINOR: Retain public constructors of classes from public API (apache#6455) KAFKA-8118; Ensure ZK clients are closed in tests, fix verification (apache#6456) KAFKA-7813: JmxTool throws NPE when --object-name is omitted KAFKA-8114: Wait for SCRAM credential propagation in DelegationTokenEndToEndAuthorizationTest (apache#6452) KAFKA-8111; Set min and max versions for Metadata requests (apache#6451) KAFKA-7855: Kafka Streams Maven Archetype quickstart fails to compile out of the box (apache#6194) MINOR: Update code to not use deprecated methods (apache#6434) MINOR: Update Trogdor ConnectionStressWorker status at the end of execution (apache#6445) KAFKA-8091; Use commitSync to check connection failure in listener update test (apache#6450) KAFKA-7027: Add an overload build method in scala (apache#6373) MINOR: Fix typos in LogValidator (apache#6449) KAFKA-7502: Cleanup KTable materialization logic in a single place (apache#6174) KAFKA-7730; Limit number of active connections per listener in brokers (KIP-402) KAFKA-8091; Remove unsafe produce from dynamic listener update test (apache#6443) MINOR: Fix JavaDocs warnings (apache#6435) MINOR: Better messaging for invalid fetch response (apache#6427) MINOR: Use Java 8 lambdas in KStreamImplTest (apache#6430)

The Java API can pass a Properties object to StreamsBuilder#build, to allow, e.g., topology optimization, while the Scala API does not yet. The latter only delegates the work to the underlying Java implementation. Reviewers: John Roesler <john@confluent.io>, Bill Bejeck <bbejeck@gmail.com>

mjsax added the streams label Mar 7, 2019

bbejeck changed the title ~~Add an overload build method in scala~~ KAFKA-7027: Add an overload build method in scala Mar 7, 2019

bbejeck reviewed Mar 7, 2019

View reviewed changes

massimosiani changed the base branch from 2.2 to trunk March 8, 2019 08:41

bbejeck approved these changes Mar 8, 2019

View reviewed changes

vvcephei requested changes Mar 8, 2019

View reviewed changes

Add an overload build method in scala

41173e3

KAFKA-7027 change the test to exclude counts

6fb9cce

bbejeck reviewed Mar 14, 2019

View reviewed changes

fix for scala 2.11

f028d31

bbejeck approved these changes Mar 14, 2019

View reviewed changes

fix an assertion

88c62df

vvcephei approved these changes Mar 15, 2019

View reviewed changes

bbejeck merged commit 853f24a into apache:trunk Mar 15, 2019

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-7027: Add an overload build method in scala #6373

KAFKA-7027: Add an overload build method in scala #6373

massimosiani commented Mar 5, 2019

massimosiani commented Mar 5, 2019

bbejeck commented Mar 7, 2019

bbejeck left a comment

bbejeck commented Mar 7, 2019

massimosiani commented Mar 8, 2019

bbejeck left a comment

bbejeck commented Mar 8, 2019

vvcephei left a comment

vvcephei Mar 8, 2019

massimosiani Mar 11, 2019

vvcephei Mar 8, 2019

massimosiani Mar 11, 2019

bbejeck Mar 13, 2019 •

edited

Loading

vvcephei Mar 8, 2019

massimosiani Mar 11, 2019

bbejeck commented Mar 13, 2019

massimosiani commented Mar 14, 2019

bbejeck left a comment •

edited

Loading

bbejeck Mar 14, 2019

massimosiani Mar 14, 2019

bbejeck Mar 14, 2019

bbejeck left a comment •

edited

Loading

bbejeck commented Mar 14, 2019

massimosiani commented Mar 15, 2019

vvcephei left a comment

bbejeck commented Mar 15, 2019

bbejeck commented Mar 15, 2019

KAFKA-7027: Add an overload build method in scala #6373

KAFKA-7027: Add an overload build method in scala #6373

Conversation

massimosiani commented Mar 5, 2019

Committer Checklist (excluded from commit message)

massimosiani commented Mar 5, 2019

bbejeck commented Mar 7, 2019

bbejeck left a comment

Choose a reason for hiding this comment

bbejeck commented Mar 7, 2019

massimosiani commented Mar 8, 2019

bbejeck left a comment

Choose a reason for hiding this comment

bbejeck commented Mar 8, 2019

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bbejeck Mar 13, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bbejeck commented Mar 13, 2019

massimosiani commented Mar 14, 2019

bbejeck left a comment • 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

bbejeck left a comment • edited Loading

Choose a reason for hiding this comment

bbejeck commented Mar 14, 2019

massimosiani commented Mar 15, 2019

vvcephei left a comment

Choose a reason for hiding this comment

bbejeck commented Mar 15, 2019

bbejeck commented Mar 15, 2019

bbejeck Mar 13, 2019 •

edited

Loading

bbejeck left a comment •

edited

Loading

bbejeck left a comment •

edited

Loading