Kafka Improvements #3189

simplesteph · 2017-04-11T03:54:16Z

Upgraded Kafka dependencies to 0.10.2,
Added dynamic configuration of producer for scala collector, which enables security configs to be passed (motivation of this PR)
Fixed configuration as well

this fixes #2973
this fixes #2974
this supersedes #3177
this fixes #3173
this fixes #2994

Fixed Kafka configuration for the Producer as well.
Added tests
Fully ready to enable security

This breaks the current recommended Kafka configuration, but as it was declared being in a "beta" state, I believe this is fine. The new config will be much more intuitive to Kafka adepts.

… scala collector, which enables security configs to be passed. Fixed configuration as well

…nstrating env variables, mandatory or optional

simplesteph · 2017-04-11T07:31:36Z

Alright done. I left a few goodies in there, including a revamp of the sample config using the true semantics from typesafe config (instead of {{ }} which definitely won't work).

Let me know your review, but I'd love to see this merged for the next release so I don't have to maintain my own Snowplow in the meantime.

The whole purpose of this was to fix a few issues and enable security configurations for Kafka

snowplowcla · 2017-04-11T08:31:41Z

@simplesteph has signed the Individual Contributor License Agreement

simplesteph · 2017-04-11T08:42:45Z

Just realized I'm addressing #2994 as well.
This jaas can be set by properties sasl.jaas.config as of kafka clients 0.10.2 (https://kafka.apache.org/documentation/#security_client_dynamicjaas)

The expected jaas is

KafkaClient {
    com.ibm.messagehub.login.MessageHubLoginModule required
    serviceName="kafka"
    username="username"
    password="password";
};

simplesteph · 2017-04-17T23:56:02Z

@alexanderdean @jbeemster have you had a chance to look at this?

simplesteph · 2017-05-25T00:18:08Z

@alexanderdean @jbeemster ping

simplesteph · 2017-08-09T05:34:58Z

@BenFradet feel free to use what you like in this for your R92, but this should cover a bunch of stuff

BenFradet · 2017-08-09T09:02:57Z

will do 👍

BenFradet

Looks good overall, please keep your changes to a minimum by only fixing the issues described in the tickets you are referencing.

BenFradet · 2017-08-10T09:31:32Z

2-collectors/scala-stream-collector/examples/config.sample.conf

-      time-limit: {{collectorSinkBufferTimeThreshold}}
+      byte-limit = 4000000
+      record-limit = 500
+      time-limit = 60000


could you leave the config as is?

BenFradet · 2017-08-10T09:31:44Z

2-collectors/scala-stream-collector/project/Dependencies.scala

@@ -33,7 +33,7 @@ object Dependencies {
    val awsSdk           = "1.6.10"
    val yodaTime         = "2.1"
    val yodaConvert      = "1.2"
-    val kafka            = "0.10.1.0"
+    val kafka            = "0.10.2.0"


this will be handled in a separate ticket

@BenFradet no not really. If you don't support this upgrade, you can't support security by passing the jaas conf as a property from producerProps (sasl.jaas.config added in 10.2)

What I mean is that there is a ticket dedicated to the upgrade and it's separate from the tickets you're trying to solve (e.g. #3325).

Adding the security feature is a separate concern from updating the library.

BenFradet · 2017-08-10T09:34:09Z

...src/main/scala/com.snowplowanalytics.snowplow.collectors/scalastream/ScalaCollectorApp.scala

+        merged
+      }
+    }
+  }


this will be overhauled in a separate ticket

that's important that it's this there if you intend to allow dynamic settings and therefore support security

BenFradet · 2017-08-10T09:34:49Z

...src/main/scala/com.snowplowanalytics.snowplow.collectors/scalastream/ScalaCollectorApp.scala

+        val props = new Properties()
+        props.put("acks", "all")
+        props.put("buffer.memory", byteLimit.toString)
+        props.put("batch.size", recordLimit.toString)


this doesn't translate, same as in the other PR

Ah true, that's bytes. byteLimit.toString then, right? And buffer.memory removed?

I don't think any current buffer setting applies to batch.size, and it doesn't make sense to have a batch.size equals to buffer.memory.

BenFradet · 2017-08-10T09:35:39Z

...r/src/main/scala/com.snowplowanalytics.snowplow.collectors/scalastream/sinks/KafkaSink.scala

-    props.put("key.serializer", 
-      "org.apache.kafka.common.serialization.StringSerializer")
-    props.put("value.serializer", 
-      "org.apache.kafka.common.serialization.ByteArraySerializer")


please keep your changes to a minimum and modify the configuration where it originally was, ie here

BenFradet · 2017-08-10T09:42:00Z

3-enrich/stream-enrich/examples/config.sample.conf

+      method = "GET"
+    }
+  }
+}


how is this different from https://github.com/snowplow/snowplow/blob/master/3-enrich/stream-enrich/examples/config.hocon.sample ?

naming a typesafe config with .sample is invalid. It messes with the compiler, and other stuff when you pass configs as arguments to your programs (especially when using docker java base image). To prevent users from making the same mistake that took me about 4 hours to solve, after debugging typesafe config itself, I'd rather rename this to config.sample.conf.

Not sure what you mean but this is the convention that has been adopted throughout all projects in Snowplow so we might as well keep things coherent.

BenFradet · 2017-08-10T09:42:20Z

3-enrich/stream-enrich/project/BuildSettings.scala

@@ -22,7 +22,7 @@ object BuildSettings {
    organization          :=  "com.snowplowanalytics",
    version               :=  "0.10.0",
    description           :=  "The Snowplow Enrichment process, implemented as an Amazon Kinesis app",
-    scalaVersion          :=  "2.10.1",
+    scalaVersion          :=  "2.10.5",


this will be handled separately

BenFradet · 2017-08-10T09:42:26Z

3-enrich/stream-enrich/project/Dependencies.scala

@@ -35,7 +35,7 @@ object Dependencies {
    val slf4j                = "1.7.5"
    val awsSdk               = "1.6.11"
    val kinesisClient        = "1.6.1"
-    val kafkaClients         = "0.10.1.0"
+    val kafkaClients         = "0.10.2.0"


BenFradet · 2017-08-10T09:43:04Z

...m-enrich/src/main/scala/com.snowplowanalytics.snowplow.enrich/kinesis/KinesisEnrichApp.scala

+    props.putAll(map.filterNot(elem => blacklist.contains(elem._1)))
+    props
+  }
+}


please modify the configuration in place

BenFradet · 2017-08-10T09:44:10Z

...m-enrich/src/test/scala/com.snowplowanalytics.snowplow.enrich.kinesis/EnrichConfigSpec.scala

+    }
+  }
+
+}


I don't think this is necesarry either

simplesteph · 2017-08-10T22:10:11Z

Hi @BenFradet , thanks for the feedback. I understand this may be more than the tickets being fixed. I don't use Snowplow anymore unfortunately, so I won't be editing this PR further. I'm happy to give you control of my branch if you want to do further commits on this branch, or you can cherry pick what you need in your own PR.
The idea was

to add dynamic config (which allows security)
properly configure Kafka defaults (minus the batch.size param)
refactor the config for extensibility.

Good luck, hope that helped!

BenFradet · 2017-08-11T08:43:52Z

@simplesteph works for me 👍 , closing.

simplesteph added 6 commits April 11, 2017 13:49

Upgraded Kafka to 0.10.2, added dynamic configuration of producer for…

8f8951b

… scala collector, which enables security configs to be passed. Fixed configuration as well

refactored collectorServiceSpec

a95a1f9

made the configuration compliant with typesafe config standards, demo…

0e803d1

…nstrating env variables, mandatory or optional

added kafka dynamic configuration for enricher. enables security

ba961c3

typo fixed

d656672

improved configuration sample, using true language from typesafe config

659fa7e

simplesteph changed the title ~~[WIP] Kafka Improvements~~ Kafka Improvements Apr 11, 2017

bump up kafka client version

3842f29

standardize syntax for sample config file

253beea

simplesteph mentioned this pull request Apr 11, 2017

fixed hocon sample file for scala-stream-collector #3188

Closed

simplesteph mentioned this pull request Jun 14, 2017

Update the wiki with Apache Kafka support #2949

Closed

BenFradet added this to the R92 Virunum (Stream refresh) milestone Aug 9, 2017

BenFradet self-assigned this Aug 10, 2017

BenFradet self-requested a review August 10, 2017 09:26

BenFradet suggested changes Aug 10, 2017

View reviewed changes

BenFradet removed this from the R92 Virunum (Stream refresh) milestone Aug 11, 2017

BenFradet removed their assignment Aug 11, 2017

BenFradet closed this Aug 11, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Kafka Improvements #3189

Kafka Improvements #3189

simplesteph commented Apr 11, 2017 •

edited

simplesteph commented Apr 11, 2017

snowplowcla commented Apr 11, 2017

simplesteph commented Apr 11, 2017

simplesteph commented Apr 17, 2017

simplesteph commented May 25, 2017

simplesteph commented Aug 9, 2017

BenFradet commented Aug 9, 2017

BenFradet left a comment

BenFradet Aug 10, 2017

BenFradet Aug 10, 2017

simplesteph Aug 10, 2017 •

edited

BenFradet Aug 11, 2017

BenFradet Aug 10, 2017

simplesteph Aug 10, 2017

BenFradet Aug 10, 2017

simplesteph Aug 10, 2017

BenFradet Aug 11, 2017

BenFradet Aug 10, 2017

BenFradet Aug 10, 2017

simplesteph Aug 10, 2017

BenFradet Aug 11, 2017

BenFradet Aug 10, 2017

BenFradet Aug 10, 2017

BenFradet Aug 10, 2017

BenFradet Aug 10, 2017

simplesteph commented Aug 10, 2017

BenFradet commented Aug 11, 2017

Kafka Improvements #3189

Kafka Improvements #3189

Conversation

simplesteph commented Apr 11, 2017 • edited

simplesteph commented Apr 11, 2017

snowplowcla commented Apr 11, 2017

simplesteph commented Apr 11, 2017

simplesteph commented Apr 17, 2017

simplesteph commented May 25, 2017

simplesteph commented Aug 9, 2017

BenFradet commented Aug 9, 2017

BenFradet left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

simplesteph Aug 10, 2017 • edited

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

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

simplesteph commented Aug 10, 2017

BenFradet commented Aug 11, 2017

simplesteph commented Apr 11, 2017 •

edited

simplesteph Aug 10, 2017 •

edited