make version non-optional

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

eventual-cassandra/src/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,

eventual-cassandra/src/test/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")),
     )

eventual-cassandra/src/test/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")),
     )

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)

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 {
@@ -18,34 +18,10 @@ object ActionHeader {
 
   implicit val formatOptActionHeader: OFormat[Option[ActionHeader]] = {
 
-    val appendFormat = {
-      val format = Json.format[Append]
-      val reads = format orElse new Reads[Append] {
-        def reads(json: JsValue) = {
-
-          def metadata = {
-            (json \ "metadata").validate[JsObject] match {
-              case JsSuccess(a, _) => a.validate[HeaderMetadata]
-              case _: JsError      => HeaderMetadata.empty.pure[JsResult]
-            }
-          }
-
-          for {
-            range       <- (json \ "range").validate[SeqRange]
-            origin      <- (json \ "origin").validateOpt[Origin]
-            version     <- (json \ "version").validateOpt[Version]
-            payloadType <- (json \ "payloadType").validate[PayloadType.BinaryOrJson]
-            metadata    <- metadata
-          } yield {
-            Append(range, origin, version, payloadType, metadata)
-          }
-        }
-      }
-      OFormat(reads, format)
-    }
-    val deleteFormat = Json.format[Delete]
-    val purgeFormat  = Json.format[Purge]
-    val readFormat   = Json.format[Mark]
+    val appendFormat = Json.using[Json.WithDefaultValues].format[Append]
+    val deleteFormat = Json.using[Json.WithDefaultValues].format[Delete]
+    val purgeFormat  = Json.using[Json.WithDefaultValues].format[Purge]
+    val readFormat   = Json.using[Json.WithDefaultValues].format[Mark]
 
     new OFormat[Option[ActionHeader]] {
 
@@ -96,25 +72,25 @@ object ActionHeader {
   final case class Append(
     range: SeqRange,
     origin: Option[Origin],
-    version: Option[Version],
+    version: Version = Version.obsolete,
     payloadType: PayloadType.BinaryOrJson,
-    metadata: HeaderMetadata,
+    metadata: HeaderMetadata = HeaderMetadata.empty,
   ) extends AppendOrDelete
 
   final case class Delete(
     to: DeleteTo,
     origin: Option[Origin],
-    version: Option[Version],
+    version: Version = Version.obsolete,
   ) extends AppendOrDelete
 
   final case class Purge(
     origin: Option[Origin],
-    version: Option[Version],
+    version: Version = Version.obsolete,
   ) extends AppendOrDelete
 
   final case class Mark(
     id: String,
     origin: Option[Origin],
-    version: Option[Version],
+    version: Version = Version.obsolete,
   ) extends ActionHeader
 }

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,
 ) {

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
       }

journal/src/test/resources/com/evolutiongaming/kafka/journal/Delete-None.json

@@ -1,6 +1,5 @@
 {
   "delete": {
-    "to": 3,
-    "version": "0.0.1"
+    "to": 3
   }
 }

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"
   }
 }

journal/src/test/resources/com/evolutiongaming/kafka/journal/Purge-origin.json

@@ -1,5 +1,6 @@
 {
   "purge": {
-    "origin": "origin"
+    "origin": "origin",
+    "version": "0.0.1"
   }
 }

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 seqNr = SeqNr.unsafe(3)
+      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")
     }
   }

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)
 

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,

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

journal/src/test/scala/com/evolutiongaming/kafka/journal/PartitionCacheSpec.scala

@@ -120,7 +120,7 @@ class PartitionCacheSpec extends AsyncFunSuite with Matchers {
   test("get HeadInfo.delete") {
     partitionCacheOf()
       .use { cache =>
-        val actionHeader = ActionHeader.Delete(to = DeleteTo(seqNr0), origin = none, version = none)
+        val actionHeader = ActionHeader.Delete(to = DeleteTo(seqNr0), origin = none, version = Version.obsolete)
         for {
           a <- cache.add(Record(id0, offset0, actionHeader))
           _ <- IO { a shouldEqual none }
@@ -638,15 +638,15 @@ object PartitionCacheSpec {
   val seqNr0: SeqNr      = SeqNr.min
   val seqNr1: SeqNr      = seqNr0.next[Try].get
 
-  val actionHeader: ActionHeader = ActionHeader.Mark("mark", none, none)
+  val actionHeader: ActionHeader = ActionHeader.Mark("mark", none, Version.obsolete)
 
   def actionHeaderOf(seqNr: SeqNr): ActionHeader.Append = {
     ActionHeader.Append(
       range       = SeqRange(seqNr),
       origin      = none,
       payloadType = PayloadType.Binary,
       metadata    = HeaderMetadata.empty,
-      version     = none,
+      version     = Version.obsolete,
     )
   }
 

journal/src/test/scala/com/evolutiongaming/kafka/journal/StreamActionRecordsSpec.scala

@@ -73,7 +73,7 @@ object StreamActionRecordsSpec {
 
     val (marker, markRecord) = {
       val offset          = pointers.lastOption.fold(1L) { _.offset + 1 }
-      val mark            = Action.Mark(key, timestamp, ActionHeader.Mark("mark", none, Version.current.some))
+      val mark            = Action.Mark(key, timestamp, ActionHeader.Mark("mark", none, Version.current))
       val partitionOffset = PartitionOffset(offset = Offset.unsafe(offset))
       val record          = ActionRecord(mark, partitionOffset)
       val marker          = Marker(mark.id, partitionOffset)
@@ -87,7 +87,7 @@ object StreamActionRecordsSpec {
       val header = ActionHeader.Append(
         range       = range,
         origin      = none,
-        version     = Version.current.some,
+        version     = Version.current,
         payloadType = PayloadType.Json,
         metadata    = HeaderMetadata.empty,
       )