#23 Create a utils class that will help users manage their properties

ABMC831 · ABMC831 · commit 7b1e5119952c · 2024-10-17T14:50:27.000+02:00
diff --git a/examples/src/main/resources/reference.conf b/examples/src/main/resources/reference.conf
@@ -0,0 +1,18 @@
+{
+  topicName = "KillMePleaseTopic"
+
+  writer {
+    bootstrap.servers = "ZADALNRAPP00009.corp.dsarena.com:9092"
+    key.serializer = "org.apache.kafka.common.serialization.StringSerializer"
+    value.serializer = "org.apache.kafka.common.serialization.StringSerializer"
+  }
+
+  reader {
+    bootstrap.servers = "ZADALNRAPP00009.corp.dsarena.com:9092"
+    key.deserializer = "org.apache.kafka.common.serialization.StringDeserializer"
+    value.deserializer = "org.apache.kafka.common.serialization.StringDeserializer"
+    client.id = "DebugConsumer_007"
+    group.id = "DebugGroup_007"
+    auto.offset.reset = "earliest"
+  }
+}
diff --git a/examples/src/main/scala-2/za.co.absa.kafkacase.examples/reader/UsingsResourceHandling.scala b/examples/src/main/scala-2/za.co.absa.kafkacase.examples/reader/UsingsResourceHandling.scala
@@ -16,11 +16,11 @@
 
 package za.co.absa.kafkacase.examples.reader
 
+import com.typesafe.config.Config
 import io.circe.Decoder
-import java.util.Properties
 
 object UsingsResourceHandling {
-  def apply[T: Decoder](readerProps: Properties, topicName: String): Unit = {
+  def apply[T: Decoder](readerConf: Config, topicName: String): Unit = {
     println("Scala 3 feature")
   }
 }
diff --git a/examples/src/main/scala-2/za.co.absa.kafkacase.examples/writer/UsingsResourceHandling.scala b/examples/src/main/scala-2/za.co.absa.kafkacase.examples/writer/UsingsResourceHandling.scala
@@ -16,11 +16,11 @@
 
 package za.co.absa.kafkacase.examples.writer
 
+import com.typesafe.config.Config
 import io.circe.Encoder
-import java.util.Properties
 
 object UsingsResourceHandling {
-  def apply[T: Encoder](writerProps: Properties, topicName: String, sampleMessageToWrite: T): Unit = {
+  def apply[T: Encoder](writerConf: Config, topicName: String, sampleMessageToWrite: T): Unit = {
     println("Scala 3 feature")
   }
 }
diff --git a/examples/src/main/scala-3/za/co/absa/kafkacase/examples/reader/UsingsResourceHandling.scala b/examples/src/main/scala-3/za/co/absa/kafkacase/examples/reader/UsingsResourceHandling.scala
@@ -19,12 +19,11 @@ package za.co.absa.kafkacase.examples.reader
 import io.circe.Decoder
 import za.co.absa.kafkacase.reader.ReaderImpl
 
-import java.util.Properties
 import scala.util.Using
 
 object UsingsResourceHandling {
-  def apply[T: Decoder](readerProps: Properties, topicName: String): Unit = {
-    Using(new ReaderImpl[T](readerProps, topicName, neverEnding = false)) { reader =>
+  def apply[T: Decoder](readerConf: Config, topicName: String): Unit = {
+    Using(new ReaderImpl[T](readerConf, topicName, neverEnding = false)) { reader =>
       for (item <- reader)
         println(item)
     }
diff --git a/examples/src/main/scala-3/za/co/absa/kafkacase/examples/writer/UsingsResourceHandling.scala b/examples/src/main/scala-3/za/co/absa/kafkacase/examples/writer/UsingsResourceHandling.scala
@@ -19,12 +19,11 @@ package za.co.absa.kafkacase.examples.writer
 import io.circe.Encoder
 import za.co.absa.kafkacase.writer.WriterImpl
 
-import java.util.Properties
 import scala.util.Using
 
 object UsingsResourceHandling {
-  def apply[T: Encoder](writerProps: Properties, topicName: String, sampleMessageToWrite: T): Unit = {
-    Using(new WriterImpl[T](writerProps, topicName)) { writer =>
+  def apply[T: Encoder](writerConf: Config, topicName: String, sampleMessageToWrite: T): Unit = {
+    Using(new WriterImpl[T](writerConf, topicName)) { writer =>
       writer.Write("sampleMessageKey1", sampleMessageToWrite)
       writer.Write("sampleMessageKey2", sampleMessageToWrite)
     }
diff --git a/examples/src/main/scala/za/co/absa/kafkacase/examples/KafkaCase.scala b/examples/src/main/scala/za/co/absa/kafkacase/examples/KafkaCase.scala
@@ -16,13 +16,12 @@
 
 package za.co.absa.kafkacase.examples
 
-import org.apache.kafka.clients.consumer.ConsumerConfig
-import org.apache.kafka.clients.producer.ProducerConfig
+import com.typesafe.config.ConfigFactory
 import za.co.absa.kafkacase.models.topics.EdlaChange
 
-import java.util.{Properties, UUID}
-
 object KafkaCase {
+  private val config = ConfigFactory.load()
+
   // This goes from your application logic
   private val sampleMessageToWrite = EdlaChange(
     app_id_snow = "N/A",
@@ -37,31 +36,12 @@ object KafkaCase {
     timestamp_event = 12345
   )
 
-  // This goes from your config / domain knowledge
-  private val topicName = "KillMePleaseTopic"
-
-  // This goes from your writer configs
-  private val writerProps = new Properties()
-  writerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "ZADALNRAPP00009.corp.dsarena.com:9092")
-  writerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
-  writerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
-
-  // This goes from your reader configs
-  private val readerProps = new Properties()
-  readerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "ZADALNRAPP00009.corp.dsarena.com:9092")
-  readerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
-  readerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
-  readerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, s"DebugConsumer_${UUID.randomUUID()}")
-  readerProps.put(ConsumerConfig.GROUP_ID_CONFIG, s"DebugGroup_${UUID.randomUUID()}")
-  readerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
-
-
   def main(args: Array[String]): Unit = {
-    writer.ManualResourceHandling(writerProps, topicName, sampleMessageToWrite)
-    writer.CustomResourceHandling(writerProps, topicName, sampleMessageToWrite)
-    writer.UsingsResourceHandling(writerProps, topicName, sampleMessageToWrite)
-    reader.ManualResourceHandling[EdlaChange](readerProps, topicName)
-    reader.CustomResourceHandling[EdlaChange](readerProps, topicName)
-    reader.UsingsResourceHandling[EdlaChange](readerProps, topicName)
+    writer.ManualResourceHandling(config.getConfig("writer"), config.getString("topicName"), sampleMessageToWrite)
+    writer.CustomResourceHandling(config.getConfig("writer"), config.getString("topicName"), sampleMessageToWrite)
+    writer.UsingsResourceHandling(config.getConfig("writer"), config.getString("topicName"), sampleMessageToWrite)
+    reader.ManualResourceHandling[EdlaChange](config.getConfig("reader"), config.getString("topicName"))
+    reader.CustomResourceHandling[EdlaChange](config.getConfig("reader"), config.getString("topicName"))
+    reader.UsingsResourceHandling[EdlaChange](config.getConfig("reader"), config.getString("topicName"))
   }
 }
diff --git a/examples/src/main/scala/za/co/absa/kafkacase/examples/reader/CustomResourceHandling.scala b/examples/src/main/scala/za/co/absa/kafkacase/examples/reader/CustomResourceHandling.scala
@@ -16,15 +16,14 @@
 
 package za.co.absa.kafkacase.examples.reader
 
+import com.typesafe.config.Config
 import io.circe.Decoder
 import za.co.absa.kafkacase.models.utils.ResourceHandler.withResource
 import za.co.absa.kafkacase.reader.ReaderImpl
 
-import java.util.Properties
-
 object CustomResourceHandling {
-  def apply[T: Decoder](readerProps: Properties, topicName: String): Unit = {
-    withResource(new ReaderImpl[T](readerProps, topicName, neverEnding = false))(reader => {
+  def apply[T: Decoder](readerConf: Config, topicName: String): Unit = {
+    withResource(new ReaderImpl[T](readerConf, topicName, neverEnding = false))(reader => {
       for (item <- reader)
         println(item)
     })
diff --git a/examples/src/main/scala/za/co/absa/kafkacase/examples/reader/ManualResourceHandling.scala b/examples/src/main/scala/za/co/absa/kafkacase/examples/reader/ManualResourceHandling.scala
@@ -16,14 +16,13 @@
 
 package za.co.absa.kafkacase.examples.reader
 
+import com.typesafe.config.Config
 import io.circe.Decoder
 import za.co.absa.kafkacase.reader.ReaderImpl
 
-import java.util.Properties
-
 object ManualResourceHandling {
-  def apply[T: Decoder](readerProps: Properties, topicName: String): Unit = {
-    val reader = new ReaderImpl[T](readerProps, topicName, neverEnding = false)
+  def apply[T: Decoder](readerConf: Config, topicName: String): Unit = {
+    val reader = new ReaderImpl[T](readerConf, topicName, neverEnding = false)
     try {
       for (item <- reader)
         println(item)
diff --git a/examples/src/main/scala/za/co/absa/kafkacase/examples/writer/CustomResourceHandling.scala b/examples/src/main/scala/za/co/absa/kafkacase/examples/writer/CustomResourceHandling.scala
@@ -16,15 +16,16 @@
 
 package za.co.absa.kafkacase.examples.writer
 
+import com.typesafe.config.Config
 import io.circe.Encoder
 import za.co.absa.kafkacase.models.utils.ResourceHandler.withResource
 import za.co.absa.kafkacase.writer.WriterImpl
 
 import java.util.Properties
 
 object CustomResourceHandling {
-  def apply[T: Encoder](writerProps: Properties, topicName: String, sampleMessageToWrite: T): Unit = {
-    withResource(new WriterImpl[T](writerProps, topicName))(writer => {
+  def apply[T: Encoder](writerConf: Config, topicName: String, sampleMessageToWrite: T): Unit = {
+    withResource(new WriterImpl[T](writerConf, topicName))(writer => {
       writer.Write("sampleMessageKey1", sampleMessageToWrite)
       writer.Write("sampleMessageKey2", sampleMessageToWrite)
     })
diff --git a/examples/src/main/scala/za/co/absa/kafkacase/examples/writer/ManualResourceHandling.scala b/examples/src/main/scala/za/co/absa/kafkacase/examples/writer/ManualResourceHandling.scala
@@ -16,14 +16,13 @@
 
 package za.co.absa.kafkacase.examples.writer
 
+import com.typesafe.config.Config
 import io.circe.Encoder
 import za.co.absa.kafkacase.writer.WriterImpl
 
-import java.util.Properties
-
 object ManualResourceHandling {
-  def apply[T: Encoder](writerProps: Properties, topicName: String, sampleMessageToWrite: T): Unit = {
-    val writer = new WriterImpl[T](writerProps, topicName)
+  def apply[T: Encoder](writerConf: Config, topicName: String, sampleMessageToWrite: T): Unit = {
+    val writer = new WriterImpl[T](writerConf, topicName)
     try {
       writer.Write("sampleMessageKey1", sampleMessageToWrite)
       writer.Write("sampleMessageKey2", sampleMessageToWrite)
diff --git a/project/Dependencies.scala b/project/Dependencies.scala
@@ -13,12 +13,14 @@ object Dependencies {
 
   def readerDependencies: Seq[ModuleID] = Seq(
     "io.circe" %% "circe-parser" % "0.14.9",
-    "org.apache.kafka" % "kafka-clients" % "3.8.0"
+    "org.apache.kafka" % "kafka-clients" % "3.8.0",
+    "com.typesafe" % "config" % "1.4.3"
   )
 
   def writerDependencies: Seq[ModuleID] = Seq(
     "io.circe" %% "circe-parser" % "0.14.9",
-    "org.apache.kafka" % "kafka-clients" % "3.8.0"
+    "org.apache.kafka" % "kafka-clients" % "3.8.0",
+    "com.typesafe" % "config" % "1.4.3"
   )
 
   def examplesDependencies: Seq[ModuleID] = Seq.empty[ModuleID]
diff --git a/reader/src/main/scala/za/co/absa/kafkacase/reader/ReaderImpl.scala b/reader/src/main/scala/za/co/absa/kafkacase/reader/ReaderImpl.scala
@@ -16,17 +16,27 @@
 
 package za.co.absa.kafkacase.reader
 
+import com.typesafe.config.Config
 import io.circe.Decoder
 import io.circe.jawn.decode
 import org.slf4j.LoggerFactory
 import org.apache.kafka.clients.consumer.{ConsumerRecord, KafkaConsumer}
-import za.co.absa.kafkacase.reader.ReaderImpl.{DEFAULT_TIMEOUT, log}
+import za.co.absa.kafkacase.reader.ReaderImpl.{DEFAULT_TIMEOUT, convertConfigToProperties, log}
 
 import java.time.Duration
 import java.util
 import java.util.Properties
 
-class ReaderImpl[TType: Decoder](props: Properties, topic: String, timeout: Duration = DEFAULT_TIMEOUT, neverEnding: Boolean = true) extends Reader[TType] {
+class ReaderImpl[TType: Decoder](props: Properties, topic: String, timeout: Duration, neverEnding: Boolean) extends Reader[TType] {
+  // note: scala can't handle default parameters together with overloading.... hence slightly exponential number of auxiliary constructors
+  def this(config: Config, topic: String, timeout: Duration) = this(convertConfigToProperties(config), topic, timeout, true)
+  def this(config: Config, topic: String, neverEnding: Boolean) = this(convertConfigToProperties(config), topic, DEFAULT_TIMEOUT, neverEnding)
+  def this(config: Config, topic: String) = this(convertConfigToProperties(config), topic, DEFAULT_TIMEOUT, true)
+  def this(config: Config, topic: String, timeout: Duration, neverEnding: Boolean) = this(convertConfigToProperties(config), topic, timeout, neverEnding)
+  def this(props: Properties, topic: String, timeout: Duration) = this(props, topic, timeout, true)
+  def this(props: Properties, topic: String, neverEnding: Boolean) = this(props, topic, DEFAULT_TIMEOUT, neverEnding)
+  def this(props: Properties, topic: String) = this(props, topic, DEFAULT_TIMEOUT, true)
+
   private val consumer = new KafkaConsumer[String, String](props)
   consumer.subscribe(util.Arrays.asList(topic))
   private var singlePollIterator = fetchNextBatch()
@@ -61,4 +71,12 @@ class ReaderImpl[TType: Decoder](props: Properties, topic: String, timeout: Dura
 object ReaderImpl {
   private val DEFAULT_TIMEOUT: Duration = Duration.ofSeconds(3)
   private val log = LoggerFactory.getLogger(this.getClass)
+
+  private def convertConfigToProperties(config: Config): Properties = {
+    val properties = new Properties()
+    config.entrySet().forEach { entry =>
+      properties.put(entry.getKey, config.getString(entry.getKey))
+    }
+    properties
+  }
 }
diff --git a/writer/src/main/scala/za/co/absa/kafkacase/writer/WriterImpl.scala b/writer/src/main/scala/za/co/absa/kafkacase/writer/WriterImpl.scala
@@ -16,15 +16,18 @@
 
 package za.co.absa.kafkacase.writer
 
+import com.typesafe.config.Config
 import io.circe.Encoder
 import io.circe.syntax.EncoderOps
 import org.slf4j.LoggerFactory
 import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
-import za.co.absa.kafkacase.writer.WriterImpl.log
+import za.co.absa.kafkacase.writer.WriterImpl.{convertConfigToProperties, log}
 
 import java.util.Properties
 
 class WriterImpl[TType: Encoder](props: Properties, topic: String) extends Writer[TType] {
+  def this(config: Config, topic: String) = this(convertConfigToProperties(config), topic)
+
   private val producer = new KafkaProducer[String, String](props)
 
   def Write(key: String, value: TType): Unit = {
@@ -39,4 +42,12 @@ class WriterImpl[TType: Encoder](props: Properties, topic: String) extends Write
 
 object WriterImpl {
   private val log = LoggerFactory.getLogger(this.getClass)
+
+  private def convertConfigToProperties(config: Config): Properties = {
+    val properties = new Properties()
+    config.entrySet().forEach { entry =>
+      properties.put(entry.getKey, config.getString(entry.getKey))
+    }
+    properties
+  }
 }

Original file line number	Diff line number	Diff line change
`@@ -16,11 +16,11 @@`
`16`	`16`
`17`	`17`	`package za.co.absa.kafkacase.examples.reader`
`18`	`18`
	`19`	`+import com.typesafe.config.Config`
`19`	`20`	`import io.circe.Decoder`
`20`		`-import java.util.Properties`
`21`	`21`
`22`	`22`	`object UsingsResourceHandling {`
`23`		`- def apply[T: Decoder](readerProps: Properties, topicName: String): Unit = {`
	`23`	`+ def apply[T: Decoder](readerConf: Config, topicName: String): Unit = {`
`24`	`24`	`println("Scala 3 feature")`
`25`	`25`	`}`
`26`	`26`	`}`