An example to start Kafka Streams

KafkaStreamsRunner.scala

val kafkaStreamsConfig: Properties = new Properties()
kafkaStreamsConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, consumerGroup)
kafkaStreamsConfig.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServers)

// Assuming streamBuilder is already defined with the transformation pipeline
val kafkaStreams = new KafkaStreams(streamBuilder, kafkaStreamsConfig)

Try {
  Logger.info("Starting Kafka Streams...")
  kafkaStreams.start()
}.recover {
  case e: Throwable =>
    Logger.error("Encountered an unrecoverable error, stopping application.", e)
    kafkaStreams.close()
    sys.exit(1)
}