Add compression, tweaked some settings

andreas-schroeder · Jan 10, 2019 · 87edcb0 · 87edcb0
1 parent 0fb6eb0
commit 87edcb0
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 5 deletions.
diff --git a/app/src/main/scala/segmenter/Segmenter.scala b/app/src/main/scala/segmenter/Segmenter.scala
@@ -39,6 +39,7 @@ class Segmenter(config: AppConfig, serdes: SegmenterSerdes) {
     PROCESSING_GUARANTEE_CONFIG                                 -> EXACTLY_ONCE,
     COMMIT_INTERVAL_MS_CONFIG                                   -> "500",
     NUM_STANDBY_REPLICAS_CONFIG                                 -> "1",
+    producerPrefix(ProducerConfig.COMPRESSION_TYPE_CONFIG)      -> "lz4",
     DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG      -> classOf[LogAndContinueExceptionHandler].getName,
     DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG                    -> classOf[SegmenterTimestampExtractor].getName,
     consumerPrefix(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG)    -> "6000",
@@ -86,8 +87,8 @@ class Segmenter(config: AppConfig, serdes: SegmenterSerdes) {
 
   // replace the event key with a session key based on the current session.
   events
-    .join(sessions)((evt, session) => (evt, session)) // session lookup.
-    .map { case (key, (evt, session)) => SessionKey(key, session.sessionId) -> evt }
+    .join(sessions)((evt, session) => (evt, session.sessionId)) // session lookup.
+    .map { case (key, (evt, sessionId)) => SessionKey(key, sessionId) -> evt }
     .to("session-events")
 
   private def printMetrics(streams: KafkaStreams): Unit = {

diff --git a/driver/src/main/scala/driver/Driver.scala b/driver/src/main/scala/driver/Driver.scala
@@ -27,13 +27,15 @@ class Driver(config: AppConfig) {
     Map(
       CLIENT_ID_CONFIG                      -> "driver",
       BOOTSTRAP_SERVERS_CONFIG              -> config.kafka.bootstrapServers,
+      ENABLE_IDEMPOTENCE_CONFIG             -> "true",
       RETRIES_CONFIG                        -> Int.MaxValue.toString,
-      MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION -> "1",
-      ACKS_CONFIG                           -> "all"
+      MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION -> "5",
+      ACKS_CONFIG                           -> "all",
+      COMPRESSION_TYPE_CONFIG               -> "lz4"
     )
   }
 
-  val keys: Seq[Device]    = (1 to 5).map(i => Device(s"device-$i"))
+  val keys: Seq[Device]    = (1 to 10).map(i => Device(s"device-$i"))
   val metrics: Seq[String] = Seq("pressure", "temperature", "wind", "humidity")
 
   val registry: String = config.kafka.schemaRegistry