Merge pull request #22 from lightbend/local-kafka-proc

The proc example application now runs with local Kafka Server

Author: debasishg

kafka-stream-q-example-dsl/README.md

@@ -83,7 +83,7 @@ $ curl http://localhost:7070/weblog/bytes/world.std.com
 
 The http query layer is designed to work even when your application runs in the distributed mode. Running your Kafka Streams application in the distributed mode means that all the instances must have the same application id.
 
-> In order to run the application in distributed mode, you need to run an external Kafka and Zookeeper server. Set `kafka.localserver` to `false` to enbale this setting.
+> In order to run the application in distributed mode, you need to run an external Kafka and Zookeeper server. Set `kafka.localserver` to `false` to enable this setting.
 
 Here are the steps that you need to follow to run the application in distributed mode. We assume here you are running both the instances in the same node with different port numbers. It's fairly easy to scale this on different nodes.
 

kafka-stream-q-example-proc/README.md

@@ -12,6 +12,8 @@ The implementation is based on the [ClarkNet dataset](http://ita.ee.lbl.gov/html
 
 ## Build and Run Locally
 
+By default the application runs through an embedded local Kafka Server. In case you want to run separate instances of Kafka and Zookeeper servers, change `kafka.localserver` to `false` in `application.conf`.
+
 To run the application, do the following steps.
 
 ### Build the Libraries
@@ -20,6 +22,8 @@ You'll need to build the Scala API library, `kafka-scala-s`, and the interactive
 
 ### Start ZooKeeper and Kafka
 
+> This is only required if the setting of `kafka.localserver` is `false` in `application.conf`. If this is set to `true`, the application runs with an embedded local Kafka server. However, note that if you want to run the application in a distributed mode(see below for details of running in distributed mode), you need to run a separate Kafka and Zookeeper server.
+
 Start ZooKeeper and Kafka, if not already running. You can download Kafka 1.0.0 for Scala 2.12 [here](https://kafka.apache.org/documentation/#quickstart), then follow the [Quick Start](https://kafka.apache.org/documentation/#quickstart) instructions for running ZooKeeper and Kafka, steps 1 and 2.
 
 ### Download the ClarkNet dataset
@@ -32,8 +36,12 @@ Copy `src/main/resources/application-proc.conf.template` to  `src/main/resources
 
 Edit `src/main/resources/application-proc.conf` and set the entry for `directorytowatch` to match the folder name where you installed the ClarkNet dataset.
 
+And note that you can run the application with a bundled local Kafka server by setting `kafka.localserver` to `true` in the `application.conf` file.
+
 ### Create the Kafka Topics
 
+> This is only required if the setting of `kafka.localserver` is `false` in `application.conf`. If this is set to `true`, the application runs with an embedded local Kafka server and creates all necessary topics on its own. However, note that if you want to run the application in a distributed mode(see below for details of running in distributed mode), you need to run a separate Kafka and Zookeeper server.
+
 Create the topics using the `kafka-topics.sh` command that comes with the Kafka distribution. We'll refer to the directory where you installed Kafka as `$KAFKA_HOME`. Run the following commands:
 
 ```bash
@@ -71,6 +79,8 @@ false
 
 The http query layer is designed to work even when your application runs in the distributed mode. Running your Kafka Streams application in the distributed mode means that all the instances must have the same application id.
 
+> In order to run the application in distributed mode, you need to run an external Kafka and Zookeeper server. Set `kafka.localserver` to `false` to enable this setting.
+
 Here are the steps that you need to follow to run the application in distributed mode. We assume here you are running both the instances in the same node with different port numbers. It's fairly easy to scale this on different nodes.
 
 ### Step 1: Build and configure for distribution

kafka-stream-q-example-proc/build.sbt

@@ -34,7 +34,9 @@ lazy val app = appProject("app")(".")
       circeGeneric,
       circeParser,
       logback,
-      scalaLogging
+      scalaLogging,
+      curator,
+      kafka
     ),
     scalacOptions ++= Seq(
       "-deprecation",                      // Emit warning and location for usages of deprecated APIs.
@@ -103,7 +105,8 @@ lazy val procRun = project
     resourceDirectory in Compile := (resourceDirectory in (app, Compile)).value,
     javaOptions in run ++= Seq(
       "-Dconfig.file=" + (resourceDirectory in Compile).value / "application-proc.conf",
-      "-Dlogback.configurationFile=" + (resourceDirectory in Compile).value / "logback-proc.xml"),
+      "-Dlogback.configurationFile=" + (resourceDirectory in Compile).value / "logback-proc.xml",
+      "-Dlog4j.configurationFile=" + (resourceDirectory in Compile).value / "log4j.properties"),
     addCommandAlias("proc", "procRun/run")
   )
   .dependsOn(app)
@@ -114,7 +117,8 @@ lazy val procPackage = appProject("procPackage")("build/proc")
     resourceDirectory in Compile := (resourceDirectory in (app, Compile)).value,
     mappings in Universal ++= {
       Seq(((resourceDirectory in Compile).value / "application-proc.conf") -> "conf/application.conf") ++
-        Seq(((resourceDirectory in Compile).value / "logback-proc.xml") -> "conf/logback.xml")
+        Seq(((resourceDirectory in Compile).value / "logback-proc.xml") -> "conf/logback.xml") ++
+        Seq(((resourceDirectory in Compile).value / "log4j.properties") -> "conf/log4j.properties")
     },
     scriptClasspath := Seq("../conf/") ++ scriptClasspath.value,
     mainClass in Compile := Some("com.lightbend.kafka.scala.iq.example.WeblogDriver")

kafka-stream-q-example-proc/project/Dependencies.scala

@@ -20,4 +20,6 @@ object Dependencies {
   val circeParser     = "io.circe"                     %% "circe-parser"             % circeVersion
   val logback         = "ch.qos.logback"                % "logback-classic"          % logbackVersion
   val scalaLogging    = "com.typesafe.scala-logging"   %% "scala-logging"            % scalaLoggingVersion
+  val curator         = "org.apache.curator"            % "curator-test"             % curatorVersion
+  val kafka           = "org.apache.kafka"             %% "kafka"                    % kafkaVersion excludeAll(ExclusionRule("org.slf4j", "slf4j-log4j12"), ExclusionRule("org.apache.zookeeper", "zookeeper")) 
 }

kafka-stream-q-example-proc/project/Versions.scala

@@ -15,5 +15,7 @@ object Versions {
   val circeVersion = "0.8.0"
   val scalaLoggingVersion = "3.5.0"
   val logbackVersion = "1.2.3"  
+  val curatorVersion = "4.0.0"
+  val kafkaVersion = "1.0.0"
 }
 

kafka-stream-q-example-proc/src/main/resources/application-proc.conf.template

@@ -6,54 +6,53 @@ akka {
   event-handlers = ["akka.event.slf4j.Slf4jEventHandler"]
 }
 
-dcos {
-
-  kafka {
-    ## bootstrap servers for Kafka
-    brokers = "localhost:9092"
-    brokers = ${?KAFKA_BROKERS}
-
-    ## consumer group
-    group = "group-proc"
-    group = ${?KAFKA_GROUP_PROC}
-
-    ## the source topic - processing starts with
-    ## data in this topic (to be loaded by ingestion)
-    fromtopic = "server-log-proc"
-    fromtopic = ${?KAFKA_FROM_TOPIC_PROC}
-
-    ## error topic for the initial processing
-    errortopic = "logerr-proc"
-    errortopic = ${?KAFKA_ERROR_TOPIC_PROC}
-
-    zookeeper = "localhost:2181"
-    zookeeper = ${?ZOOKEEPER_URL}
-
-    ## folder where state stores are created by Kafka Streams
-    statestoredir = "/tmp/kafka-streams"
-    statestoredir = ${?STATESTOREDIR}
-
-    ## settings for data ingestion
-    loader {
-      sourcetopic = ${dcos.kafka.fromtopic}
-      sourcetopic = ${?KAFKA_FROM_TOPIC_PROC}
-
-      directorytowatch = "/Users/myhome/ClarkNet-HTTP"
-      directorytowatch = ${?DIRECTORY_TO_WATCH}
-
-      pollinterval = 1 second
-    }
+kafka {
+  # true if use local kafka server
+  # false otherwise
+  # if true, then setting of brokers below is ignored and set to that of KafkaLocalServer
+  localserver = true
+
+  ## bootstrap servers for Kafka
+  brokers = "localhost:9092"
+  brokers = ${?KAFKA_BROKERS}
+
+  ## consumer group
+  group = "group-proc"
+  group = ${?KAFKA_GROUP_PROC}
+
+  ## the source topic - processing starts with
+  ## data in this topic (to be loaded by ingestion)
+  fromtopic = "server-log-proc"
+  fromtopic = ${?KAFKA_FROM_TOPIC_PROC}
+
+  ## error topic for the initial processing
+  errortopic = "logerr-proc"
+  errortopic = ${?KAFKA_ERROR_TOPIC_PROC}
+
+  ## folder where state stores are created by Kafka Streams
+  statestoredir = "/tmp/kafka-streams"
+  statestoredir = ${?STATESTOREDIR}
+
+  ## settings for data ingestion
+  loader {
+    sourcetopic = ${kafka.fromtopic}
+    sourcetopic = ${?KAFKA_FROM_TOPIC_PROC}
+
+    directorytowatch = "/Users/myhome/ClarkNet-HTTP"
+    directorytowatch = ${?DIRECTORY_TO_WATCH}
+
+    pollinterval = 1 second
   }
+}
 
-  # http endpoints of the weblog microservice
-  http {
-    # The port the dashboard listens on
-    port = 7071
-    port = ${?PORT0}
+# http endpoints of the weblog microservice
+http {
+  # The port the dashboard listens on
+  port = 7071
+  port = ${?PORT0}
 
-    # The interface the dashboard listens on
-    interface = "localhost"
-    interface = ${?INTERFACE_PROC}
-  }
+  # The interface the dashboard listens on
+  interface = "localhost"
+  interface = ${?INTERFACE_PROC}
 }
 

kafka-stream-q-example-proc/src/main/resources/log4j.properties

@@ -0,0 +1,16 @@
+# Set root logger level to DEBUG and its only appender to A1.
+log4j.rootLogger=ERROR, R
+
+# A1 is set to be a ConsoleAppender.
+log4j.appender.A1=org.apache.log4j.ConsoleAppender
+
+log4j.appender.R=org.apache.log4j.RollingFileAppender
+log4j.appender.R.File=logs/kafka-server.log
+
+log4j.appender.R.MaxFileSize=100KB
+# Keep one backup file
+log4j.appender.R.MaxBackupIndex=1
+
+# A1 uses PatternLayout.
+log4j.appender.R.layout=org.apache.log4j.PatternLayout
+log4j.appender.R.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n

kafka-stream-q-example-proc/src/main/scala/com/lightbend/kafka/scala/iq/example/WeblogWorkflow.scala

@@ -21,6 +21,7 @@ import serializers._
 import com.lightbend.kafka.scala.iq.http.InteractiveQueryHttpService
 
 import ingestion.DataIngestion
+import com.lightbend.kafka.scala.server._
 
 trait WeblogWorkflow extends LazyLogging with AppSerializers {
 
@@ -33,6 +34,7 @@ trait WeblogWorkflow extends LazyLogging with AppSerializers {
     }
 
     logger.info(s"config = $config")
+    val maybeServer = startLocalServerIfSetInConfig(config)
 
     // setup REST endpoints
     val restEndpointPort = config.httpPort
@@ -41,6 +43,7 @@ trait WeblogWorkflow extends LazyLogging with AppSerializers {
 
     logger.info("Connecting to Kafka cluster via bootstrap servers " + config.brokers)
     logger.warn("REST endpoint at http://" + restEndpointHostName + ":" + restEndpointPort)
+    println("Connecting to Kafka cluster via bootstrap servers " + config.brokers)
     println("REST endpoint at http://" + restEndpointHostName + ":" + restEndpointPort)
 
     implicit val system = ActorSystem()
@@ -87,6 +90,8 @@ trait WeblogWorkflow extends LazyLogging with AppSerializers {
         case x: Exception => x.printStackTrace
       } finally {
         logger.error("Exiting application ..")
+        logger.error(s"Stopping kafka server ..")
+        maybeServer.foreach(_.stop())
         System.exit(-1)
       }
     })
@@ -101,11 +106,24 @@ trait WeblogWorkflow extends LazyLogging with AppSerializers {
       restService.stop()
       val closed = streams.close(1, TimeUnit.MINUTES)
       logger.error(s"Exiting application after streams close ($closed)")
+      maybeServer.foreach(_.stop())
     } catch {
       case _: Exception => // ignored
     }))
   }  
 
+  private def createTopics(config: ConfigData, server: KafkaLocalServer) = {
+    import config._
+    List(fromTopic, errorTopic).foreach(server.createTopic(_))
+  }
+
+  private def startLocalServerIfSetInConfig(config: ConfigData): Option[KafkaLocalServer] = if (config.localServer) {
+    val s = KafkaLocalServer(true, Some(config.stateStoreDir))
+    s.start()
+    createTopics(config, s)
+    Some(s)
+  } else None 
+
   def createStreams(config: ConfigData): KafkaStreams
   def startRestProxy(streams: KafkaStreams, hostInfo: HostInfo,
     actorSystem: ActorSystem, materializer: ActorMaterializer): InteractiveQueryHttpService

kafka-stream-q-example-proc/src/main/scala/com/lightbend/kafka/scala/iq/example/config/KStreamConfig.scala

@@ -8,6 +8,7 @@ import cats.instances.all._
 import scala.util.Try
 import com.typesafe.config.Config
 import scala.concurrent.duration._
+import com.lightbend.kafka.scala.server._
 
 
 /**
@@ -22,8 +23,8 @@ object KStreamConfig {
   )
 
   private[KStreamConfig] case class ServerSettings(
+    localServer: Boolean,
     brokers: String, 
-    zk: String, 
     stateStoreDir: String
   )
 
@@ -44,8 +45,8 @@ object KStreamConfig {
   )
 
   case class ConfigData(ks: KafkaSettings, hs: HttpSettings, dls: DataLoaderSettings) {
+    def localServer = ks.serverSettings.localServer
     def brokers = ks.serverSettings.brokers
-    def zk = ks.serverSettings.zk
     def fromTopic = ks.topicSettings.fromTopic
     def errorTopic = ks.topicSettings.errorTopic
     def stateStoreDir = ks.serverSettings.stateStoreDir
@@ -67,15 +68,26 @@ object KStreamConfig {
 
   private def fromKafkaConfig: ConfigReader[KafkaSettings] = Kleisli { (config: Config) =>
     Try {
+      val local = config.getBoolean("kafka.localserver")
+      val serverSettings = 
+        if (local) {
+          ServerSettings(
+            local,
+            s"localhost:${KafkaLocalServer.DefaultPort}",
+            config.getString("kafka.statestoredir")
+          )
+        } else {
+          ServerSettings(
+            local,
+            config.getString("kafka.brokers"),
+            config.getString("kafka.statestoredir")
+          )
+        }
       KafkaSettings(
-        ServerSettings(
-          config.getString("dcos.kafka.brokers"),
-          config.getString("dcos.kafka.zookeeper"),
-          config.getString("dcos.kafka.statestoredir")
-        ),
+        serverSettings,
         TopicSettings(
-          config.getString("dcos.kafka.fromtopic"),
-          config.getString("dcos.kafka.errortopic")
+          config.getString("kafka.fromtopic"),
+          config.getString("kafka.errortopic")
         )
       )
     }
@@ -84,18 +96,18 @@ object KStreamConfig {
   private def fromHttpConfig: ConfigReader[HttpSettings] = Kleisli { (config: Config) =>
     Try {
       HttpSettings(
-        config.getString("dcos.http.interface"),
-        config.getInt("dcos.http.port")
+        config.getString("http.interface"),
+        config.getInt("http.port")
       )
     }
   }
 
   private def fromDataLoaderConfig: ConfigReader[DataLoaderSettings] = Kleisli { (config: Config) =>
     Try {
       DataLoaderSettings(
-        config.getString("dcos.kafka.loader.sourcetopic"),
-        getStringMaybe(config, "dcos.kafka.loader.directorytowatch"),
-        config.getDuration("dcos.kafka.loader.pollinterval")
+        config.getString("kafka.loader.sourcetopic"),
+        getStringMaybe(config, "kafka.loader.directorytowatch"),
+        config.getDuration("kafka.loader.pollinterval")
       )
     }
   }