Introducing support for Kafka streams #39

adamosloizou · 2016-10-17T20:42:50Z

Description

Hello,

I have been using your library recently and I really like it 😃!
I am currently working with Kafka's new library, Kafka Streams and I couldn't find an easy abstraction for testing.

This PR introduces a ScalaTest-based trait that should make testing Kafka Streams testing easier.

Features/Changes

Codebase is now multimodule. Moved existing code base to embedded-kafka and introduced kafka-streams submodule. This allows for people to not have to depend on Kafka Streams for using the original code base.
The new module now offers EmbeddedKafkaStreams and EmbeddedKafkaStreamsAllInOne traits. They use and expand on the EmbeddedKafkatrait.
The design uses the loaner-pattern for safe resource management for creating and destroying Kafka streams as well as test consumers
It adds a few useful tools to the existing code base, particularly Consumers and ConsumerExtensions for easily consuming arbitrary messages as Scala Streams.

Testing

Tests on kafka-streams pass
Tests on existing code base embedded-kafka get [Address already in use errors] which I also see in master. Is that expected?

Notes

I have not changed the README.md as I would like to see how you like it before discussing what the user manual should look like.

Thanks!
Adamos

manub · 2016-10-17T21:40:20Z

Hi @adamosloizou and thanks for your PR! I'll go through it in the next days - however I think your changes are incompatible with Scala 2.10 (there's a transitive dependency missing in Travis build for 2.10). I'm not sure whether we need to keep cross versions compatibility, but if you have an easy way to fix that it would be appreciated.

Will keep you posted on the rest of the PR (I haven't used Kafka Streams myself).

adamosloizou · 2016-10-17T21:45:18Z

Hey @manub! Yes, it was Typesafe's scalalogging which I have removed so hopefully it should be ok for Scala 2.10 as well.

As for Kafka Streams, interesting concept but its API is still young. You can definitely spot commonalities between that and Spark and Akka Streams API for functional-style transformations (maps, flatmaps etc). It's a nice way forward it seems.

Let me know what you think!

manub · 2016-10-18T16:03:41Z

Try to merge master in your branch and run again the build please.

adamosloizou · 2016-10-21T10:29:50Z

@manub I think I fixed it. There was a clash with the two sub-modules' tests using the same ports for their Kafka/Zookeeper servers.

adamosloizou · 2016-10-21T20:40:37Z

This is really weird, it works fine locally on my machine. I am running with JVM 8 instead of 7 as the build. Could that have something to do?

manub · 2016-10-23T21:36:07Z

I will have a look at it - had a failure with my Mac and I'm in the middle of reinstalling all the software. Sorry for the delay!

adamosloizou · 2016-10-24T14:26:48Z

I have managed to replicate the issue on my machine by using a VM with limited resources. It was a race-condition that exposed the fact that the Stream consumers could start consuming after the 1st set of published messages.
I fixed it by configuring the streams' consumers to start at the earliest message.

However, now I am seeing a failure on the embedded-kafka side. Any ideas?
Thanks.

manub · 2016-10-24T15:44:11Z

Rerunning your build passes it - it's quite unstable on CI and I'm aware of that. Will look at the code as soon as I have an opportunity. As I mentioned before my personal laptop failed and I'm still in the process of restoring it. Thanks for your patience!

manub

Hey @adamosloizou - thanks for the PR! It looks overall good although there are some coding style things that I'd like to get fixed before merging. Also - would it be possible for you to extend the README.md?

manub · 2016-10-24T20:29:52Z

embedded-kafka/src/main/scala/net/manub/embeddedkafka/ConsumerExtensions.scala

+      */
+    def consumeLazily(topic: String): Stream[(K, V)] = {
+      val attempts = 1 to MaximumAttempts
+      attempts.toStream.flatMap(attempt => {


could this be rewritten as .flatMap { attempt => ... }?

Fixed in badc3b2

manub · 2016-10-24T20:32:19Z

embedded-kafka/src/main/scala/net/manub/embeddedkafka/UUIDs.scala

+    *
+    * @return the unique ID
+    */
+  def newUuid(): String = UUID.randomUUID().toString


I'd rather have this method still returning a UUID and let the user convert it to string.

Fixed in badc3b2

manub · 2016-10-24T20:33:19Z

embedded-kafka/src/main/scala/net/manub/embeddedkafka/Consumers.scala

+  def withConsumer[K: Deserializer, V: Deserializer, T](block: KafkaConsumer[K, V] => T)
+                                                       (implicit config: EmbeddedKafkaConfig): T = {
+    val consumer = newConsumer[K, V]()
+    val result = block(consumer)


can block() ever throw an exception? as probably we don't know in advance it would be better to wrap block(consumer) into a try / finally block in order to close the consumer regardless of what happens.

Good catch!
Fixed in badc3b2

manub · 2016-10-24T20:33:44Z

embedded-kafka/src/test/resources/log4j.properties

@@ -1,7 +1,8 @@
-log4j.rootLogger=info, stdout
+log4j.rootLogger=off
+#log4j.rootLogger=info, stdout


this line can be removed

Fixed in 2cace27

manub · 2016-10-24T20:34:22Z

kafka-streams/src/main/scala/net/manub/embeddedkafka/streams/EmbeddedKafkaStreams.scala

+                (block: => Any)
+                (implicit config: EmbeddedKafkaConfig): Any =
+    withRunningKafka {
+      topicsToCreate.foreach(topic => createCustomTopic(topic))


in my opinion it would be more idiomatic .foreach(createCustomTopic)

I agree. I tried that originally but unfortunately it does not work:

Error:(45, 30) type mismatch; found : (String, Map[String,String], Int, Int) => Unit required: String => ? topicsToCreate.foreach(createCustomTopic)

An alternative is foreach(createCustomTopic(_)). In my opinion that's not particularly pretty and I prefer being explicit as per above. I think it reads better.
What do you think?

manub · 2016-10-24T20:34:51Z

kafka-streams/src/main/scala/net/manub/embeddedkafka/streams/EmbeddedKafkaStreams.scala

+                (implicit config: EmbeddedKafkaConfig): Any =
+    withRunningKafka {
+      topicsToCreate.foreach(topic => createCustomTopic(topic))
+      val streamId: String = UUIDs.newUuid()


this type annotation it's redundand

Fixed in badc3b2

manub · 2016-10-24T20:35:34Z

kafka-streams/src/test/resources/log4j.properties

@@ -0,0 +1,9 @@
+log4j.rootLogger=off
+#log4j.logger.net.manub.embeddedkafka=debug, stdout
+#log4j.logger.org.apache.kafka.streams=debug, stdout


those 2 lines can be removed as well

Fixed in 2cace27

manub · 2016-10-24T20:39:32Z

kafka-streams/src/test/scala/net/manub/embeddedkafka/streams/ExampleKafkaStreamsSpec.scala

+import org.apache.kafka.streams.kstream.{KStream, KStreamBuilder}
+import org.scalatest.{FlatSpec, Matchers}
+
+class ExampleKafkaStreamsSpec extends FlatSpec with Matchers with EmbeddedKafkaStreamsAllInOne {


Would it be possible to use WordSpec here for consistency reasons?

Of course!
Fixed in badc3b2

adamosloizou · 2016-10-25T22:55:13Z

README updated 😄!

…ggregation project in preparation for introducing kafka-streams subproject.

…ded Kafka. Introduced EmbeddedKafkaStreams and EmbeddedKafkaStreamsAllInOne traits for easy testing. Introduced Consumers trait for easy creation and consumption of test messages.

… tests in build.

adamosloizou · 2016-10-26T00:13:22Z

@manub Please re-run the build. I think the failures are the unstable part that you mentioned.

manub · 2016-10-26T21:30:16Z

Thanks @adamosloizou !

adamosloizou force-pushed the kafka-streams branch 2 times, most recently from 92e615f to 30bc5ea Compare October 17, 2016 21:38

adamosloizou force-pushed the kafka-streams branch from 30bc5ea to 4116932 Compare October 21, 2016 10:28

adamosloizou force-pushed the kafka-streams branch from 1cf6fe3 to fae64c5 Compare October 24, 2016 14:00

manub requested changes Oct 24, 2016

View reviewed changes

manub mentioned this pull request Oct 25, 2016

Switch to Kafka 0.10.1.0 #40

Merged

adamosloizou force-pushed the kafka-streams branch from fae64c5 to 2cace27 Compare October 25, 2016 22:50

adamosloizou and others added 3 commits October 26, 2016 00:56

Moved embedded kafka to its own sub module. Changed root project to a…

66353f7

…ggregation project in preparation for introducing kafka-streams subproject.

Introduced kafka-streams module for testing Kafka Streams using Embed…

9d04fb6

…ded Kafka. Introduced EmbeddedKafkaStreams and EmbeddedKafkaStreamsAllInOne traits for easy testing. Introduced Consumers trait for easy creation and consumption of test messages.

Disabling verbose logging for tests to make it easier to trace failed…

e661bdc

… tests in build.

adamosloizou force-pushed the kafka-streams branch from 2cace27 to e661bdc Compare October 26, 2016 00:03

manub approved these changes Oct 26, 2016

View reviewed changes

manub merged commit c30918a into manub:master Oct 26, 2016

Introducing support for Kafka streams #39

Introducing support for Kafka streams #39

Uh oh!

Conversation

adamosloizou commented Oct 17, 2016

Description

Features/Changes

Testing

Notes

Uh oh!

manub commented Oct 17, 2016

Uh oh!

adamosloizou commented Oct 17, 2016

Uh oh!

manub commented Oct 18, 2016

Uh oh!

adamosloizou commented Oct 21, 2016

Uh oh!

adamosloizou commented Oct 21, 2016

Uh oh!

manub commented Oct 23, 2016

Uh oh!

adamosloizou commented Oct 24, 2016

Uh oh!

manub commented Oct 24, 2016

Uh oh!

manub left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

adamosloizou commented Oct 25, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

adamosloizou commented Oct 26, 2016

Uh oh!

manub commented Oct 26, 2016

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

adamosloizou commented Oct 25, 2016 •

edited

Loading