Feature/#15 sync schema with event gate (#16)

ABMC831 · web-flow · commit f5961585daa6 · 2024-09-20T09:02:08.000+02:00
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
@@ -28,16 +28,16 @@ object KafkaCase {
   private def writer_use_case(): Unit = {
     // 0 -> HAVE SOMETHING TO WRITE
     val messageToWrite = EdlaChangeTopic(
-      id = "DebugId",
       app_id_snow = "N/A",
-      source_app = "ThisCode",
+      data_definition_id = "TestingThis",
       environment = "DEV",
-      timestamp_event = 12345,
-      data_definition = "TestingThis",
-      operation = EdlaChangeTopic.Operation.CREATE(),
-      location = "ether",
       format = "FooBar",
-      schema_link = "http://not.here"
+      guid = "DebugId",
+      location = "ether",
+      operation = EdlaChangeTopic.Operation.CREATE(),
+      schema_link = "http://not.here",
+      source_app = "ThisCode",
+      timestamp_event = 12345
     )
 
     // 1 -> DEFINE PROPS - kafka to treat all as string
diff --git a/models/src/main/scala/za/co/absa/KafkaCase/Models/EdlaChangeTopic.scala b/models/src/main/scala/za/co/absa/KafkaCase/Models/EdlaChangeTopic.scala
@@ -21,16 +21,16 @@ import io.circe.generic.JsonCodec
 
 @JsonCodec
 case class EdlaChangeTopic(
-  id: String,
   app_id_snow: String,
-  source_app: String,
+  data_definition_id: String,
   environment: String,
-  timestamp_event: Long,
-  data_definition: String,
-  operation: EdlaChangeTopic.Operation,
-  location: String,
   format: String,
-  schema_link: String
+  guid: String,
+  location: String,
+  operation: EdlaChangeTopic.Operation,
+  schema_link: String,
+  source_app: String,
+  timestamp_event: Long
 )
 
 object EdlaChangeTopic {
diff --git a/models/src/main/scala/za/co/absa/KafkaCase/Models/SchemaRunTopic.scala b/models/src/main/scala/za/co/absa/KafkaCase/Models/SchemaRunTopic.scala
@@ -21,16 +21,16 @@ import io.circe.generic.JsonCodec
 
 @JsonCodec
 case class SchemaRunTopic(
-  id: String,
-  job_ref: String,
   app_id_snow: String,
-  source_app: String,
-  environment: String,
-  timestamp_start: Long,
-  timestamp_end: Long,
   data_definition_id: String,
+  environment: String,
+  guid: String,
+  job_ref: String,
+  message: String,
+  source_app: String,
   status: SchemaRunTopic.Status,
-  message: String
+  timestamp_end: Long,
+  timestamp_start: Long
 )
 
 object SchemaRunTopic {
diff --git a/models/src/test/scala/za/co/absa/KafkaCase/Models/EdlaChangeTopicUnitTests.scala b/models/src/test/scala/za/co/absa/KafkaCase/Models/EdlaChangeTopicUnitTests.scala
@@ -14,39 +14,38 @@
  * limitations under the License.
  */
 
-package za.co.absa.KafkaCase.models
+package za.co.absa.KafkaCase.Models
 
 import io.circe.jawn.decode
 import io.circe.syntax.EncoderOps
 import org.scalatest.funsuite.AnyFunSuite
-import za.co.absa.KafkaCase.Models.EdlaChangeTopic
 
 class EdlaChangeTopicUnitTests extends AnyFunSuite {
   private val instance =  EdlaChangeTopic(
-    id = "DebugId",
     app_id_snow = "N/A",
-    source_app = "ThisCode",
+    data_definition_id = "TestingThis",
     environment = "DEV",
-    timestamp_event = 12345,
-    data_definition = "TestingThis",
-    operation = EdlaChangeTopic.Operation.CREATE(),
-    location = "ether",
     format = "FooBar",
-    schema_link = "http://not.here"
+    guid = "DebugId",
+    location = "ether",
+    operation = EdlaChangeTopic.Operation.CREATE(),
+    schema_link = "http://not.here",
+    source_app = "ThisCode",
+    timestamp_event = 12345
   )
 
   private val json =
     """{
-      |  "id" : "DebugId",
       |  "app_id_snow" : "N/A",
-      |  "source_app" : "ThisCode",
+      |  "data_definition_id" : "TestingThis",
       |  "environment" : "DEV",
-      |  "timestamp_event" : 12345,
-      |  "data_definition" : "TestingThis",
-      |  "operation" : "CREATE",
-      |  "location" : "ether",
       |  "format" : "FooBar",
-      |  "schema_link" : "http://not.here"
+      |  "guid" : "DebugId",
+      |  "location" : "ether",
+      |  "operation" : "CREATE",
+      |  "schema_link" : "http://not.here",
+      |  "source_app" : "ThisCode",
+      |  "timestamp_event" : 12345
       |}""".stripMargin
 
   test("Serializes to JSON properly") {
diff --git a/models/src/test/scala/za/co/absa/KafkaCase/Models/SchemaRunTopicUnitTests.scala b/models/src/test/scala/za/co/absa/KafkaCase/Models/SchemaRunTopicUnitTests.scala
@@ -14,39 +14,38 @@
  * limitations under the License.
  */
 
-package za.co.absa.KafkaCase.models
+package za.co.absa.KafkaCase.Models
 
 import io.circe.jawn.decode
 import io.circe.syntax.EncoderOps
 import org.scalatest.funsuite.AnyFunSuite
-import za.co.absa.KafkaCase.Models.SchemaRunTopic
 
 class SchemaRunTopicUnitTests extends AnyFunSuite {
   private val instance =  SchemaRunTopic(
-    id = "DebugId",
-    job_ref = "UnitTestJob",
     app_id_snow = "N/A",
-    source_app = "ThisTest",
-    environment = "TEST",
-    timestamp_start = 12345,
-    timestamp_end = 67890,
     data_definition_id = "Foo",
+    environment = "TEST",
+    guid = "DebugId",
+    job_ref = "UnitTestJob",
+    message = "FooBar",
+    source_app = "ThisTest",
     status = SchemaRunTopic.Status.Killed(),
-    message = "FooBar"
+    timestamp_end = 67890,
+    timestamp_start = 12345
   )
 
   private val json =
     """{
-      |  "id" : "DebugId",
-      |  "job_ref" : "UnitTestJob",
       |  "app_id_snow" : "N/A",
-      |  "source_app" : "ThisTest",
-      |  "environment" : "TEST",
-      |  "timestamp_start" : 12345,
-      |  "timestamp_end" : 67890,
       |  "data_definition_id" : "Foo",
+      |  "environment" : "TEST",
+      |  "guid" : "DebugId",
+      |  "job_ref" : "UnitTestJob",
+      |  "message" : "FooBar",
+      |  "source_app" : "ThisTest",
       |  "status" : "Killed",
-      |  "message" : "FooBar"
+      |  "timestamp_end" : 67890,
+      |  "timestamp_start" : 12345
       |}""".stripMargin
 
   test("Serializes to JSON properly") {
diff --git a/reader/src/main/scala/za/co/absa/KafkaCase/Reader/Reader.scala b/reader/src/main/scala/za/co/absa/KafkaCase/Reader/Reader.scala
@@ -16,4 +16,4 @@
 
 package za.co.absa.KafkaCase.Reader
 
-trait Reader[TType] extends Iterator[(String, TType)] with AutoCloseable
+trait Reader[TType] extends Iterator[(String, Either[String, TType])] with AutoCloseable
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
@@ -33,13 +33,16 @@ class ReaderImpl[TType: Decoder](props: Properties, topic: String, timeout: Dura
 
   override def hasNext: Boolean = singlePollIterator.hasNext
 
-  override def next(): (String, TType) = {
+  override def next(): (String, Either[String, TType]) = {
     log.info("Fetching next item")
     val nextItem = singlePollIterator.next()
-    val nextItemTyped = decode[TType](nextItem.value()).getOrElse(throw new Exception(s"Cannot parse $nextItem"))
     if (!singlePollIterator.hasNext)
       singlePollIterator = fetchNextBatch()
-    nextItem.key() -> nextItemTyped
+    val nextItemMaybeTyped = decode[TType](nextItem.value()) match {
+      case Left(_) => Left(s"Cannot parse ${nextItem.value()}")
+      case Right(item) => Right(item)
+    }
+    nextItem.key() -> nextItemMaybeTyped
   }
 
   def close(): Unit = consumer.close()

Original file line number	Diff line number	Diff line change
`@@ -16,4 +16,4 @@`
`16`	`16`
`17`	`17`	`package za.co.absa.KafkaCase.Reader`
`18`	`18`
`19`		`-trait Reader[TType] extends Iterator[(String, TType)] with AutoCloseable`
	`19`	`+trait Reader[TType] extends Iterator[(String, Either[String, TType])] with AutoCloseable`