remove obsolete benchmark unit

evolution-gaming · Nov 6, 2024 · 0744f0d · 0744f0d
1 parent 74382ca
commit 0744f0d
Show file tree

Hide file tree

Showing 25 changed files with 142 additions and 81 deletions.
diff --git a/core/src/main/scala/com/evolutiongaming/kafka/journal/Version.scala b/core/src/main/scala/com/evolutiongaming/kafka/journal/Version.scala
@@ -14,6 +14,9 @@ object Version {
 
   val current: Version = Version(Option(Version.getClass.getPackage.getImplementationVersion).getOrElse("unknown"))
 
+  /** The last release before [[Version]] was introduced, should be used only as fallback during data recovery */
+  val obsolete: Version = Version("0.0.152")
+
   implicit val eqVersion: Eq[Version] = Eq.fromUniversalEquals
 
   implicit val showVersion: Show[Version] = Show.fromToString

diff --git a/...c/main/scala/com/evolutiongaming/kafka/journal/eventual/cassandra/JournalStatements.scala b/...c/main/scala/com/evolutiongaming/kafka/journal/eventual/cassandra/JournalStatements.scala
@@ -122,7 +122,7 @@ private[journal] object JournalStatements {
             .encode(record.event.partitionOffset)
             .encode("timestamp", record.event.timestamp)
             .encodeSome(record.event.origin)
-            .encodeSome(record.event.version)
+            .encode(record.event.version)
             .encode("tags", record.event.event.tags)
             .encodeSome("meta_record_id", record.metaRecordId)
             .encodeSome("payload_type", payloadType)
@@ -189,7 +189,7 @@ private[journal] object JournalStatements {
       } yield {
         new SelectRecords[F] {
 
-          def apply(key: Key, segment: SegmentNr, range: SeqRange) = {
+          def apply(key: Key, segment: SegmentNr, range: SeqRange): Stream[F, JournalRecord] = {
 
             def readPayload(row: Row): Option[EventualPayloadAndType] = {
               val payloadType = row.decode[Option[PayloadType]]("payload_type")
@@ -226,7 +226,7 @@ private[journal] object JournalStatements {
                 event           = event,
                 timestamp       = row.decode[Instant]("timestamp"),
                 origin          = row.decode[Option[Origin]],
-                version         = row.decode[Option[Version]],
+                version         = row.decode[Option[Version]].getOrElse(Version.obsolete),
                 partitionOffset = partitionOffset,
                 metadata        = metadata,
                 headers         = headers,

diff --git a/...st/scala/com/evolutiongaming/kafka/journal/eventual/cassandra/EventualCassandraTest.scala b/...st/scala/com/evolutiongaming/kafka/journal/eventual/cassandra/EventualCassandraTest.scala
@@ -42,7 +42,7 @@ class EventualCassandraTest extends AnyFunSuite with Matchers {
       timestamp       = timestamp0,
       partitionOffset = partitionOffset,
       origin          = origin.some,
-      version         = version.some,
+      version         = version,
       metadata        = RecordMetadata(HeaderMetadata(Json.obj(("key", "value")).some), PayloadMetadata.empty),
       headers         = Headers(("key", "value")),
     )

diff --git a/.../scala/com/evolutiongaming/kafka/journal/eventual/cassandra/ReplicatedCassandraTest.scala b/.../scala/com/evolutiongaming/kafka/journal/eventual/cassandra/ReplicatedCassandraTest.scala
@@ -47,7 +47,7 @@ class ReplicatedCassandraTest extends AnyFunSuite with Matchers {
       timestamp       = timestamp0,
       partitionOffset = partitionOffset,
       origin          = origin.some,
-      version         = version.some,
+      version         = version,
       metadata        = RecordMetadata(HeaderMetadata(Json.obj(("key", "value")).some), PayloadMetadata.empty),
       headers         = Headers(("key", "value")),
     )

diff --git a/journal/src/main/scala/com/evolutiongaming/kafka/journal/Action.scala b/journal/src/main/scala/com/evolutiongaming/kafka/journal/Action.scala
@@ -17,7 +17,7 @@ sealed abstract class Action extends Product {
 
   def origin: Option[Origin] = header.origin
 
-  def version: Option[Version] = header.version
+  def version: Version = header.version
 }
 
 object Action {
@@ -79,7 +79,7 @@ object Action {
       key: Key,
       timestamp: Instant,
       origin: Option[Origin],
-      version: Option[Version],
+      version: Version,
       events: Events[A],
       metadata: HeaderMetadata,
       headers: Headers,
@@ -120,7 +120,7 @@ object Action {
       timestamp: Instant,
       to: DeleteTo,
       origin: Option[Origin],
-      version: Option[Version],
+      version: Version,
     ): Delete = {
       val header = ActionHeader.Delete(to, origin, version)
       Delete(key, timestamp, header)
@@ -135,7 +135,7 @@ object Action {
 
   object Purge {
 
-    def apply(key: Key, timestamp: Instant, origin: Option[Origin], version: Option[Version]): Purge = {
+    def apply(key: Key, timestamp: Instant, origin: Option[Origin], version: Version): Purge = {
       val header = ActionHeader.Purge(origin, version)
       Purge(key, timestamp, header)
     }
@@ -157,7 +157,7 @@ object Action {
       timestamp: Instant,
       id: String,
       origin: Option[Origin],
-      version: Option[Version],
+      version: Version,
     ): Mark = {
       val header = ActionHeader.Mark(id, origin, version)
       Mark(key, timestamp, header)

diff --git a/journal/src/main/scala/com/evolutiongaming/kafka/journal/ActionHeader.scala b/journal/src/main/scala/com/evolutiongaming/kafka/journal/ActionHeader.scala
@@ -9,7 +9,7 @@ sealed abstract class ActionHeader extends Product {
 
   def origin: Option[Origin]
 
-  def version: Option[Version]
+  def version: Version
 }
 
 object ActionHeader {
@@ -21,7 +21,7 @@ object ActionHeader {
     val appendFormat = {
       val format = Json.format[Append]
       val reads = format orElse new Reads[Append] {
-        def reads(json: JsValue) = {
+        def reads(json: JsValue): JsResult[Append] = {
 
           def metadata = {
             (json \ "metadata").validate[JsObject] match {
@@ -37,15 +37,57 @@ object ActionHeader {
             payloadType <- (json \ "payloadType").validate[PayloadType.BinaryOrJson]
             metadata    <- metadata
           } yield {
-            Append(range, origin, version, payloadType, metadata)
+            Append(range, origin, version.getOrElse(Version.obsolete), payloadType, metadata)
+          }
+        }
+      }
+      OFormat(reads, format)
+    }
+    val deleteFormat = {
+      val format = Json.format[Delete]
+      val reads = format orElse new Reads[Delete] {
+        def reads(json: JsValue): JsResult[Delete] = {
+          for {
+            to      <- (json \ "to").validate[SeqNr]
+            origin  <- (json \ "origin").validateOpt[Origin]
+            version <- (json \ "version").validateOpt[Version]
+          } yield {
+            Delete(to.toDeleteTo, origin, version.getOrElse(Version.obsolete))
+          }
+        }
+      }
+      OFormat(reads, format)
+    }
+    val purgeFormat = {
+
+      val format = Json.format[Purge]
+      val reads = format orElse new Reads[Purge] {
+        def reads(json: JsValue): JsResult[Purge] = {
+          for {
+            origin  <- (json \ "origin").validateOpt[Origin]
+            version <- (json \ "version").validateOpt[Version]
+          } yield {
+            Purge(origin, version.getOrElse(Version.obsolete))
+          }
+        }
+      }
+      OFormat(reads, format)
+    }
+    val readFormat = {
+      val format = Json.format[Mark]
+      val reads = format orElse new Reads[Mark] {
+        def reads(json: JsValue): JsResult[Mark] = {
+          for {
+            id      <- (json \ "id").validate[String]
+            origin  <- (json \ "origin").validateOpt[Origin]
+            version <- (json \ "version").validateOpt[Version]
+          } yield {
+            Mark(id, origin, version.getOrElse(Version.obsolete))
           }
         }
       }
       OFormat(reads, format)
     }
-    val deleteFormat = Json.format[Delete]
-    val purgeFormat  = Json.format[Purge]
-    val readFormat   = Json.format[Mark]
 
     new OFormat[Option[ActionHeader]] {
 
@@ -96,25 +138,25 @@ object ActionHeader {
   final case class Append(
     range: SeqRange,
     origin: Option[Origin],
-    version: Option[Version],
+    version: Version,
     payloadType: PayloadType.BinaryOrJson,
     metadata: HeaderMetadata,
   ) extends AppendOrDelete
 
   final case class Delete(
     to: DeleteTo,
     origin: Option[Origin],
-    version: Option[Version],
+    version: Version,
   ) extends AppendOrDelete
 
   final case class Purge(
     origin: Option[Origin],
-    version: Option[Version],
+    version: Version,
   ) extends AppendOrDelete
 
   final case class Mark(
     id: String,
     origin: Option[Origin],
-    version: Option[Version],
+    version: Version,
   ) extends ActionHeader
 }
diff --git a/journal/src/main/scala/com/evolutiongaming/kafka/journal/EventRecord.scala b/journal/src/main/scala/com/evolutiongaming/kafka/journal/EventRecord.scala
@@ -14,7 +14,7 @@ final case class EventRecord[A](
   timestamp: Instant,
   partitionOffset: PartitionOffset,
   origin: Option[Origin],
-  version: Option[Version],
+  version: Version,
   metadata: RecordMetadata,
   headers: Headers,
 ) {

diff --git a/journal/src/main/scala/com/evolutiongaming/kafka/journal/Produce.scala b/journal/src/main/scala/com/evolutiongaming/kafka/journal/Produce.scala
@@ -90,7 +90,7 @@ private[journal] object Produce {
             ActionHeader.Append(
               range       = range,
               origin      = origin,
-              version     = version.some,
+              version     = version,
               payloadType = payloadAndType.payloadType,
               metadata    = metadata,
             ),
@@ -109,15 +109,15 @@ private[journal] object Produce {
       def delete(key: Key, to: DeleteTo): F[PartitionOffset] = {
         for {
           timestamp <- Clock[F].instant
-          action     = Action.Delete(key, timestamp, to, origin, version.some)
+          action     = Action.Delete(key, timestamp, to, origin, version)
           result    <- send(action)
         } yield result
       }
 
       def purge(key: Key): F[PartitionOffset] = {
         for {
           timestamp <- Clock[F].instant
-          action     = Action.Purge(key, timestamp, origin, version.some)
+          action     = Action.Purge(key, timestamp, origin, version)
           result    <- send(action)
         } yield result
       }
@@ -126,7 +126,7 @@ private[journal] object Produce {
         for {
           timestamp <- Clock[F].instant
           id         = randomId.value
-          action     = Action.Mark(key, timestamp, id, origin, version.some)
+          action     = Action.Mark(key, timestamp, id, origin, version)
           result    <- send(action)
         } yield result
       }

diff --git a/journal/src/test/resources/com/evolutiongaming/kafka/journal/Delete-None.json b/journal/src/test/resources/com/evolutiongaming/kafka/journal/Delete-None.json
@@ -1,6 +1,5 @@
 {
   "delete": {
-    "to": 3,
-    "version": "0.0.1"
+    "to": 3
   }
 }
diff --git a/journal/src/test/resources/com/evolutiongaming/kafka/journal/Mark-origin.json b/journal/src/test/resources/com/evolutiongaming/kafka/journal/Mark-origin.json
@@ -1,6 +1,7 @@
 {
   "mark": {
     "id": "id",
-    "origin": "origin"
+    "origin": "origin",
+    "version": "0.0.1"
   }
 }
diff --git a/journal/src/test/resources/com/evolutiongaming/kafka/journal/Purge-origin.json b/journal/src/test/resources/com/evolutiongaming/kafka/journal/Purge-origin.json
@@ -1,5 +1,6 @@
 {
   "purge": {
-    "origin": "origin"
+    "origin": "origin",
+    "version": "0.0.1"
   }
 }
diff --git a/journal/src/test/scala/com/evolutiongaming/kafka/journal/ActionHeaderJsonSpec.scala b/journal/src/test/scala/com/evolutiongaming/kafka/journal/ActionHeaderJsonSpec.scala
@@ -30,24 +30,42 @@ class ActionHeaderJsonSpec extends AnyFunSuite with Matchers {
       test(s"Append format, origin: $origin, payloadType: $payloadType, metadata: $metadataStr") {
         val range = SeqRange.unsafe(1, 5)
         val header =
-          ActionHeader.Append(range = range, origin = origin, version = none, payloadType = payloadType, metadata = metadata)
+          ActionHeader.Append(
+            range       = range,
+            origin      = origin,
+            version     = Version.obsolete,
+            payloadType = payloadType,
+            metadata    = metadata,
+          )
         verify(header, s"Append-$originStr-$payloadType-$metadataStr")
       }
     }
 
     test(s"Delete format, origin: $origin") {
       val seqNr  = SeqNr.unsafe(3)
-      val header = ActionHeader.Delete(seqNr.toDeleteTo, origin, Version("0.0.1").some)
+      val version = origin match {
+        case Some(_) => Version("0.0.1")
+        case None => Version.obsolete
+      }
+      val header = ActionHeader.Delete(seqNr.toDeleteTo, origin, version)
       verify(header, s"Delete-$originStr")
     }
 
     test(s"Purge format, origin: $origin") {
-      val header = ActionHeader.Purge(origin, none)
+      val version = origin match {
+        case Some(_) => Version("0.0.1")
+        case None => Version.obsolete
+      }
+      val header = ActionHeader.Purge(origin, version)
       verify(header, s"Purge-$originStr")
     }
 
     test(s"Mark format, origin: $origin") {
-      val header = ActionHeader.Mark("id", origin, none)
+      val version = origin match {
+        case Some(_) => Version("0.0.1")
+        case None => Version.obsolete
+      }
+      val header = ActionHeader.Mark("id", origin, version)
       verify(header, s"Mark-$originStr")
     }
   }

diff --git a/journal/src/test/scala/com/evolutiongaming/kafka/journal/ActionToProducerRecordSpec.scala b/journal/src/test/scala/com/evolutiongaming/kafka/journal/ActionToProducerRecordSpec.scala
@@ -29,7 +29,7 @@ class ActionToProducerRecordSpec extends AnyFunSuite with Matchers {
 
   private val origins = List(Origin("origin").some, none[Origin])
 
-  private val versions = List(Version.current.some, none[Version])
+  private val versions = List(Version.current)
 
   private val seqNrs = List(SeqNr.min, SeqNr.max)
 

diff --git a/journal/src/test/scala/com/evolutiongaming/kafka/journal/HeadCacheSpec.scala b/journal/src/test/scala/com/evolutiongaming/kafka/journal/HeadCacheSpec.scala
@@ -109,7 +109,7 @@ class HeadCacheSpec extends AsyncWordSpec with Matchers {
             result <- Concurrent[IO].start { headCache.get(key = key, partition = partition, offset = marker) }
             _      <- stateRef.update { _.copy(topics = Map((topic, List(partition)))) }
             _ <- stateRef.update { state =>
-              val action  = Action.Mark(key, timestamp, ActionHeader.Mark("mark", none, Version.current.some))
+              val action  = Action.Mark(key, timestamp, ActionHeader.Mark("mark", none, Version.current))
               val record  = consumerRecordOf(action, topicPartition, marker)
               val records = ConsumerRecordsOf(List(record))
               state.enqueue(records.pure[Try])
@@ -260,7 +260,7 @@ object HeadCacheSpec {
         key       = key,
         timestamp = timestamp,
         origin    = none,
-        version   = Version.current.some,
+        version   = Version.current,
         events    = Events(Nel.of(Event(seqNr)), PayloadMetadata.empty),
         metadata  = recordMetadata,
         headers   = headers,

diff --git a/journal/src/test/scala/com/evolutiongaming/kafka/journal/HeadInfoSpec.scala b/journal/src/test/scala/com/evolutiongaming/kafka/journal/HeadInfoSpec.scala
@@ -66,20 +66,20 @@ class HeadInfoSpec extends AnyFunSuite with Matchers {
     ActionHeader.Append(
       range       = SeqRange.unsafe(from, to),
       origin      = None,
-      version     = Version.current.some,
+      version     = Version.current,
       payloadType = PayloadType.Json,
       metadata    = HeaderMetadata.empty,
     )
   }
 
   private def delete(seqNr: Int) = {
     val deleteTo = SeqNr.unsafe(seqNr).toDeleteTo
-    ActionHeader.Delete(deleteTo, none, Version.current.some)
+    ActionHeader.Delete(deleteTo, none, Version.current)
   }
 
-  private def mark = ActionHeader.Mark("id", none, Version.current.some)
+  private def mark = ActionHeader.Mark("id", none, Version.current)
 
-  private def purge = ActionHeader.Purge(none, Version.current.some)
+  private def purge = ActionHeader.Purge(none, Version.current)
 
   private def deleteInfo(seqNr: Int) = {
     val deleteTo = SeqNr.unsafe(seqNr).toDeleteTo