Feature/improve configurations and error handling (#20)

* set scalac flags

* refactoring and custom exception

Author: davideicardi

KaaSchemaRegistry/src/main/scala/com/davideicardi/kaa/KaaSchemaRegistry.scala

@@ -5,10 +5,7 @@ import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
 import org.apache.avro.SchemaNormalization
 import org.apache.kafka.clients.consumer.KafkaConsumer
 import java.util.{Collections, Properties, UUID}
-import java.util.concurrent.Executors
-import java.util.concurrent.ExecutorService
 import java.time.{Duration => JavaDuration}
-import collection.JavaConverters._
 import com.github.blemale.scaffeine.{ Cache, Scaffeine }
 import org.apache.kafka.clients.producer.ProducerConfig
 import org.apache.kafka.clients.consumer.ConsumerConfig
@@ -20,6 +17,7 @@ import java.util.concurrent.atomic.AtomicBoolean
 import scala.concurrent.Future
 import scala.concurrent.ExecutionContext
 import scala.concurrent.Await
+import com.davideicardi.kaa.utils.RetryConfig
 
 object KaaSchemaRegistry {
   val DEFAULT_TOPIC_NAME = "schemas-v1"
@@ -30,20 +28,18 @@ class KaaSchemaRegistry(
   topic: String = KaaSchemaRegistry.DEFAULT_TOPIC_NAME,
   cliendId: String = "KaaSchemaRegistry",
   pollInterval: Duration = 5.second,
-  getRetries: Int = 5,
-  getRetryDelay: Duration = 2.second
+  getRetry: RetryConfig = RetryConfig(5, 2.second)
 ) extends SchemaRegistry {
 
-  // TODO Eval to put this code inside an "init" function instead of here in the constructor
-  private val producer: KafkaProducer[java.lang.Long, String]
-    = new KafkaProducer(createProducerConfig(), new LongSerializer(), new StringSerializer())
-  private val consumer: KafkaConsumer[java.lang.Long, String]
-    = new KafkaConsumer(createConsumerConfig(), new LongDeserializer(), new StringDeserializer())
-  private var executor: ExecutorService = null
+  implicit private val ec = ExecutionContext.global
+
+  private val producer = createProducer()
+  private val consumer = createConsumer()
   private val cache: Cache[Long, String] = Scaffeine().build[Long, String]()
   private val stopping = new AtomicBoolean(false)
-  implicit private val ec = ExecutionContext.global
-  private val subscriber = Future {
+  private val startConsumerFuture = startConsumer()
+
+  private def startConsumer(): Future[Unit] = Future {
     consumer.subscribe(Collections.singletonList(topic))
       val jPollInterval = JavaDuration.ofMillis(pollInterval.toMillis)
       while (!stopping.get()) {
@@ -57,13 +53,14 @@ class KaaSchemaRegistry(
       consumer.close();
   }
 
-  // TODO eval if shutdown is called properly
   def shutdown(): Unit = {
     stopping.set(true)
-    Await.result(subscriber, 10.seconds)
+    Await.result(startConsumerFuture, 10.seconds)
   }
 
   override def put(schema: Schema): SchemaId = {
+    if (stopping.get()) throw new UnsupportedOperationException("KaaSchemaRegistry is not available")
+
     val fingerprint = SchemaNormalization.parsingFingerprint64(schema)
 
     if (cache.getIfPresent(fingerprint).isEmpty) {
@@ -75,28 +72,36 @@ class KaaSchemaRegistry(
   }
 
   override def get(id: SchemaId): Option[Schema] = {
-    Retry.retryIfNone(getRetries, getRetryDelay) {
+    Retry.retryIfNone(getRetry) {
       cache.getIfPresent(id.value)
         .map(new Schema.Parser().parse)
     }
   }
 
-  def createConsumerConfig(): Properties = {
+  protected def createConsumer() = {
+    new KafkaConsumer(consumerProps(), new LongDeserializer(), new StringDeserializer())    
+  }
+
+  protected def consumerProps(): Properties = {
     val props = new Properties()
     props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
     props.put(ConsumerConfig.CLIENT_ID_CONFIG, cliendId)
     props.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString());
     props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
     props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
-    // TODO study if we need other default properties and allow to extend this from outside
+
     props
   }
 
-  def createProducerConfig(): Properties = {
+  protected def createProducer() = {
+    new KafkaProducer(producerProps(), new LongSerializer(), new StringSerializer())    
+  }
+
+  protected def producerProps(): Properties = {
     val props = new Properties()
     props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
     props.put(ProducerConfig.CLIENT_ID_CONFIG, cliendId)
-    // TODO study if we need other default properties and allow to extend this from outside
+ 
     props
   }
 }

KaaSchemaRegistry/src/main/scala/com/davideicardi/kaa/KaaSchemaRegistryAdmin.scala

@@ -1,18 +1,7 @@
 package com.davideicardi.kaa
 
-import org.apache.avro.Schema
-import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
-import org.apache.avro.SchemaNormalization
-import org.apache.kafka.clients.consumer.KafkaConsumer
-import java.util.{Collections, Properties, UUID, Optional}
-import java.util.concurrent.Executors
-import java.util.concurrent.ExecutorService
+import java.util.{Collections, Properties, Optional}
 import scala.jdk.CollectionConverters._
-import com.github.blemale.scaffeine.{ Cache, Scaffeine }
-import org.apache.kafka.clients.producer.ProducerConfig
-import org.apache.kafka.clients.consumer.ConsumerConfig
-import org.apache.kafka.common.serialization.{LongDeserializer, StringDeserializer}
-import org.apache.kafka.common.serialization.{LongSerializer, StringSerializer}
 import org.apache.kafka.clients.admin.{AdminClient, AdminClientConfig, NewTopic}
 import org.apache.kafka.common.config.TopicConfig
 
@@ -32,7 +21,7 @@ class KaaSchemaRegistryAdmin(
     )
     newTopic.configs(newTopicsConfigs.asJava)
 
-    adminClient.createTopics(Collections.singletonList(newTopic)).all().get()
+    val _ = adminClient.createTopics(Collections.singletonList(newTopic)).all().get()
   }
 
   def topicExists(): Boolean = {
@@ -41,6 +30,6 @@ class KaaSchemaRegistryAdmin(
   }
 
   def deleteTopic(): Unit = {
-    adminClient.deleteTopics(Collections.singletonList(topic)).all().get()
+    val _ = adminClient.deleteTopics(Collections.singletonList(topic)).all().get()
   }
 }

KaaSchemaRegistry/src/main/scala/com/davideicardi/kaa/SchemaRegistry.scala

@@ -1,18 +1,6 @@
 package com.davideicardi.kaa
 
 import org.apache.avro.Schema
-import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
-import org.apache.avro.SchemaNormalization
-import org.apache.kafka.clients.consumer.KafkaConsumer
-import java.util.{Collections, Properties}
-import java.util.concurrent.Executors
-import java.util.concurrent.ExecutorService
-import collection.JavaConverters._
-import com.github.blemale.scaffeine.{ Cache, Scaffeine }
-import org.apache.kafka.clients.producer.ProducerConfig
-import org.apache.kafka.clients.consumer.ConsumerConfig
-import org.apache.kafka.common.serialization.{LongDeserializer, StringDeserializer}
-import org.apache.kafka.common.serialization.{LongSerializer, StringSerializer}
 
 trait SchemaRegistry {
   def put(schema: Schema): SchemaId

KaaSchemaRegistry/src/main/scala/com/davideicardi/kaa/avro/AvroSingleObjectEncoding.scala

@@ -3,6 +3,7 @@ package com.davideicardi.kaa.avro
 import java.nio.{ByteBuffer, ByteOrder}
 
 import com.davideicardi.kaa.SchemaId
+import com.davideicardi.kaa.InvalidSchemaException
 
 /**
  * https://avro.apache.org/docs/current/spec.html#single_object_encoding
@@ -24,7 +25,7 @@ object AvroSingleObjectEncoding {
       SchemaId(id) -> data
     }
     else {
-      throw new Exception(s"Byte array is not in correct format. First ${V1_HEADER.length} bytes are not equal" +
+      throw new InvalidSchemaException(s"Byte array is not in correct format. First ${V1_HEADER.length} bytes are not equal" +
         s" to ${V1_HEADER.mkString("[", ", ", "]")}")
     }
   }

KaaSchemaRegistry/src/main/scala/com/davideicardi/kaa/avro/AvroSingleObjectSerializer.scala

@@ -2,6 +2,7 @@ package com.davideicardi.kaa.avro
 
 import com.sksamuel.avro4s._
 import com.davideicardi.kaa.SchemaRegistry
+import com.davideicardi.kaa.SchemaNotFoundException
 
 class AvroSingleObjectSerializer[T >: Null : SchemaFor : Encoder : Decoder]
 (schemaRegistry: SchemaRegistry){
@@ -18,7 +19,7 @@ class AvroSingleObjectSerializer[T >: Null : SchemaFor : Encoder : Decoder]
   def deserialize(bytes: Array[Byte]): T = {
     val (schemaId, serialized) = AvroSingleObjectEncoding.decode(bytes)
     val schema = schemaRegistry.get(schemaId)
-      .getOrElse(throw new Exception(s"Schema $schemaId not found in registry"))
+      .getOrElse(throw new SchemaNotFoundException(s"Schema $schemaId not found in registry"))
 
     binarySerializer.read(serialized, schema)
   }

KaaSchemaRegistry/src/main/scala/com/davideicardi/kaa/exceptions.scala

@@ -0,0 +1,6 @@
+package com.davideicardi.kaa
+
+
+class InvalidSchemaException(s:String) extends Exception(s){}
+
+class SchemaNotFoundException(s:String) extends Exception(s){}

KaaSchemaRegistry/src/main/scala/com/davideicardi/kaa/utils/Retry.scala

@@ -5,19 +5,24 @@ import scala.concurrent._
 
 object Retry {
     @annotation.tailrec
-    def retryIfNone[T](count: Int, delay: Duration)(fn: => Option[T]): Option[T] = {
+    def retryIfNone[T](retryConfig: RetryConfig)(fn: => Option[T]): Option[T] = {
         fn match {
             case Some(x) => Some(x)
-            case None if count > 1 => {
-                Sleep.sleep(delay.toMillis)
-                println("retrying...")
-                retryIfNone(count - 1, delay)(fn)
+            case None if retryConfig.count > 1 => {
+                Sleep.sleep(retryConfig.delay.toMillis)
+                retryIfNone(retryConfig.copy(count = retryConfig.count - 1))(fn)
             }
             case None => None
         }
     }
 }
 
+object RetryConfig {
+  val NoRetry = RetryConfig(0, 0.second)
+}
+
+case class RetryConfig(count: Int, delay: Duration) {}
+
 object Sleep {
   def sleep(millis: Long): Unit =
     try {

KaaSchemaRegistry/src/test/scala/kaa/darwin/AvroSingleObjectSerializerSpec.scala

@@ -36,9 +36,6 @@ class AvroSingleObjectSerializerSpec extends AnyFlatSpec with should.Matchers {
     bin should be (binarySerializer.write(dragonite))
   }
 
-  // TODO add tests to verify that it is backward and forward compatibile
-  // add tests to verify what's happening when schema is not found
-
   object AvroUtils {
     def calcFingerprint(schema: Schema): SchemaId = {
       SchemaId(SchemaNormalization.parsingFingerprint64(schema))

SampleApp/src/main/scala/sampleApp.scala

@@ -1,6 +1,5 @@
 import com.davideicardi.kaa.avro.AvroSingleObjectSerializer
 import com.davideicardi.kaa.KaaSchemaRegistry
-import scala.tools.nsc.doc.html.HtmlTags.Sup
 import com.davideicardi.kaa.KaaSchemaRegistryAdmin
 object SampleApp {
     def main(args: Array[String]): Unit = {

build.sbt

@@ -3,7 +3,38 @@ import xerial.sbt.Sonatype._
 // Common settings
 organization in ThisBuild := "com.davideicardi"
 scalaVersion in ThisBuild := "2.13.3"
-scalacOptions in ThisBuild += "-deprecation"
+scalacOptions in ThisBuild ++= Seq(
+  "-deprecation", // Emit warning and location for usages of deprecated APIs.
+  "-explaintypes", // Explain type errors in more detail.
+  "-feature", // Emit warning and location for usages of features that should be imported explicitly.
+  "-unchecked", // Enable additional warnings where generated code depends on assumptions.
+  "-Xcheckinit", // Wrap field accessors to throw an exception on uninitialized access.
+  "-Xfatal-warnings", // Fail the compilation if there are any warnings.
+  "-Xlint:adapted-args", // Warn if an argument list is modified to match the receiver.
+  "-Xlint:constant", // Evaluation of a constant arithmetic expression results in an error.
+  "-Xlint:delayedinit-select", // Selecting member of DelayedInit.
+  "-Xlint:doc-detached", // A Scaladoc comment appears to be detached from its element.
+  "-Xlint:inaccessible", // Warn about inaccessible types in method signatures.
+  "-Xlint:infer-any", // Warn when a type argument is inferred to be `Any`.
+  "-Xlint:missing-interpolator", // A string literal appears to be missing an interpolator id.
+  "-Xlint:nullary-unit", // Warn when nullary methods return Unit.
+  "-Xlint:option-implicit", // Option.apply used implicit view.
+  "-Xlint:package-object-classes", // Class or object defined in package object.
+  "-Xlint:poly-implicit-overload", // Parameterized overloaded implicit methods are not visible as view bounds.
+  "-Xlint:private-shadow", // A private field (or class parameter) shadows a superclass field.
+  "-Xlint:stars-align", // Pattern sequence wildcard must align with sequence component.
+  "-Xlint:type-parameter-shadow", // A local type parameter shadows a type already in scope.
+  "-Ywarn-dead-code", // Warn when dead code is identified.
+  "-Ywarn-extra-implicit", // Warn when more than one implicit parameter section is defined.
+  "-Ywarn-numeric-widen", // Warn when numerics are widened.
+  "-Ywarn-unused:implicits", // Warn if an implicit parameter is unused.
+  "-Ywarn-unused:imports", // Warn if an import selector is not referenced.
+  "-Ywarn-unused:locals", // Warn if a local definition is unused.
+  "-Ywarn-unused:params", // Warn if a value parameter is unused.
+  "-Ywarn-unused:patvars", // Warn if a variable bound in a pattern is unused.
+  "-Ywarn-unused:privates", // Warn if a private member is unused.
+  "-Ywarn-value-discard", // Warn when non-Unit expression results are unused.
+)
 
 // sbt-dynver version settings
 dynverSonatypeSnapshots in ThisBuild := true