make version non-optional

Author: mr-git

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