spline #1155 Fix ArangoJack deserialization of generic types

arangodb-foxx-api/pom.xml

@@ -97,5 +97,9 @@
             <groupId>io.swagger</groupId>
             <artifactId>swagger-annotations</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-core</artifactId>
+        </dependency>
     </dependencies>
 </project>

arangodb-foxx-api/src/main/scala/za/co/absa/spline/consumer/service/model/Frame.scala

@@ -16,8 +16,6 @@
 
 package za.co.absa.spline.consumer.service.model
 
-import java.{util => ju}
-
 /**
  * Represents a frame of items with pagination information.
  *
@@ -27,7 +25,7 @@ import java.{util => ju}
  * @tparam T type of items
  */
 case class Frame[T](
-  items: ju.List[T],
+  items: Seq[T],
   totalCount: Long,
   offset: Int
 )

arangodb-foxx-api/src/main/scala/za/co/absa/spline/persistence/model/entities.scala

@@ -15,6 +15,8 @@
  */
 package za.co.absa.spline.persistence.model
 
+import com.fasterxml.jackson.core.`type`.TypeReference
+
 trait ArangoDocument {
   val _key: ArangoDocument.Key = null // NOSONAR
 
@@ -92,6 +94,8 @@ case class DataSource(
 }
 
 object DataSource {
+  implicit val typeRef: TypeReference[DataSource] = new TypeReference[DataSource] {}
+
   type Key = ArangoDocument.Key
   type Uri = String
   type Name = String

commons/pom.xml

@@ -56,6 +56,14 @@
             <optional>true</optional>
         </dependency>
 
+        <!-- package: jackson -->
+
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-core</artifactId>
+            <optional>true</optional>
+        </dependency>
+
         <!-- package: validation -->
 
         <dependency>

commons/src/main/scala/com/fasterxml/jackson/core/type/package.scala

@@ -0,0 +1,24 @@
+/*
+ * Copyright 2024 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.fasterxml.jackson.core
+
+package object `type` {
+  object TypeReference {
+    implicit val typeRefBoolean: TypeReference[Boolean] = new TypeReference[Boolean] {}
+    implicit val typeRefUnit: TypeReference[Nothing] = new TypeReference[Nothing] {}
+  }
+}

consumer-services/src/main/scala/za/co/absa/spline/consumer/service/model/LineageOverview.scala

@@ -15,6 +15,7 @@
  */
 package za.co.absa.spline.consumer.service.model
 
+import com.fasterxml.jackson.core.`type`.TypeReference
 import io.swagger.annotations.{ApiModel, ApiModelProperty}
 
 @ApiModel(description = "Lineage/Impact Overview")
@@ -24,3 +25,7 @@ case class LineageOverview(
   @ApiModelProperty(value = "Additional information")
   info: Map[String, Any]
 )
+
+object LineageOverview {
+  implicit val typeRef: TypeReference[LineageOverview] = new TypeReference[LineageOverview] {}
+}

persistence/src/main/scala/com/arangodb/util/TypeRefAwareArangoDeserializerDecor.scala

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2024 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.arangodb.util
+
+import com.arangodb.velocypack.VPackSlice
+import com.fasterxml.jackson.core.`type`.TypeReference
+import com.fasterxml.jackson.databind.ObjectMapper
+import za.co.absa.commons.reflect.ReflectionUtils
+
+import java.lang.reflect.Type
+
+/**
+ * This is a workaround for a ArangoJack deserializer limitation.
+ * The ArangoJack serde being a wrapper over the Jackson serde needs to be able to take a `TypeReference` instance
+ * and pass it over to Jackson in order to deserialize generic entities. This is not possible in the current ArangoJack version.
+ * Hence, we have to add another method that we'd use to pass the type reference instance to the ArrangoJack's underlying Jackson
+ * mapper directly, bypassing the ArangoJack's public API.
+ */
+class TypeRefAwareArangoDeserializerDecor(deserializer: ArangoDeserializer) extends ArangoDeserializer {
+  private lazy val vpackMapper: ObjectMapper = ReflectionUtils.extractValue[ObjectMapper](deserializer, "vpackMapper")
+
+  def deserialize[A](vpack: VPackSlice)(implicit typeRef: TypeReference[A]): A = {
+    val aType = typeRef.getType
+    if (aType == classOf[String]) {
+      deserializer.deserialize[A](vpack, aType)
+    } else {
+      vpackMapper.readValue(vpack.getBuffer, vpack.getStart, vpack.getStart + vpack.getByteSize, typeRef)
+    }
+  }
+
+  override def deserialize[T](vpack: VPackSlice, aType: Type): T = deserializer.deserialize[T](vpack, aType)
+}

persistence/src/main/scala/za/co/absa/spline/persistence/FoxxRouter.scala

@@ -19,43 +19,43 @@ package za.co.absa.spline.persistence
 import com.arangodb.ArangoDBException
 import com.arangodb.async.ArangoDatabaseAsync
 import com.arangodb.internal.util.ArangoSerializationFactory.Serializer
+import com.arangodb.util.{ArangoSerializer, TypeRefAwareArangoDeserializerDecor}
+import com.fasterxml.jackson.core.`type`.TypeReference
 import org.springframework.beans.factory.annotation.Autowired
 
 import java.util.concurrent.CompletionException
 import scala.PartialFunction.cond
 import scala.concurrent.{ExecutionContext, Future}
 import scala.jdk.FutureConverters._
-import scala.reflect.ClassTag
 
 class FoxxRouter @Autowired()(db: ArangoDatabaseAsync) {
-  private val serialization = db.util(Serializer.CUSTOM)
+  private val serializer: ArangoSerializer = db.util(Serializer.CUSTOM)
+  private val deserializer = new TypeRefAwareArangoDeserializerDecor(db.util(Serializer.CUSTOM))
 
-  def get[A: ClassTag](endpoint: String, queryParams: Map[String, Any] = Map.empty)(implicit ex: ExecutionContext): Future[A] = {
-    val aType = implicitly[ClassTag[A]].runtimeClass
+  def get[A](endpoint: String, queryParams: Map[String, Any] = Map.empty)(implicit ex: ExecutionContext, typeRef: TypeReference[A]): Future[A] = {
     val routeBuilder = db.route(endpoint)
 
     for ((k, v) <- queryParams)
       routeBuilder.withQueryParam(k, v)
 
     routeBuilder.get()
       .asScala
-      .map(resp => serialization.deserialize[A](resp.getBody, aType))
+      .map(resp => deserializer.deserialize(resp.getBody))
       .recover({
         case ce: CompletionException
           if cond(ce.getCause)({ case ae: ArangoDBException => ae.getResponseCode == 404 }) =>
           throw new NoSuchElementException(s"Resource NOT FOUND: $endpoint")
       })
   }
 
-  def post[A: ClassTag](endpoint: String, body: AnyRef)(implicit ex: ExecutionContext): Future[A] = {
-    val aType = implicitly[ClassTag[A]].runtimeClass
-    val serializedBody = serialization.serialize(body)
+  def post[A](endpoint: String, body: AnyRef)(implicit ex: ExecutionContext, typeRef: TypeReference[A]): Future[A] = {
+    val serializedBody = serializer.serialize(body)
     db
       .route(endpoint)
       .withBody(serializedBody)
       .post()
       .asScala
-      .map(resp => serialization.deserialize[A](resp.getBody, aType))
+      .map(resp => deserializer.deserialize(resp.getBody))
       .asInstanceOf[Future[A]]
   }
 }

persistence/src/test/scala/com/arangodb/util/TypeRefAwareArangoDeserializerDecorSpec.scala

@@ -0,0 +1,100 @@
+/*
+ * Copyright 2024 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.arangodb.util
+
+import com.arangodb.mapping.ArangoJack
+import com.arangodb.util.TypeRefAwareArangoDeserializerDecorSpec._
+import com.arangodb.velocypack.{VPackParser, VPackSlice}
+import com.fasterxml.jackson.core.`type`.TypeReference
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.mockito.Mockito.{verify, verifyNoMoreInteractions}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import org.scalatestplus.mockito.MockitoSugar.mock
+
+import java.lang.reflect.Type
+import scala.reflect.ClassTag
+
+class TypeRefAwareArangoDeserializerDecorSpec extends AnyFlatSpec with Matchers {
+
+  private val vpackParser = new VPackParser.Builder().build
+
+  behavior of "deserialize(VPackSlice, Type) method"
+
+  it should "delegate to underlying ArangoJack deserializer unconditionally" in {
+    val dummyVPack = mock[VPackSlice]
+    val dummyType = mock[Type]
+    val mockArangoDeserializer = mock[ArangoDeserializer]
+    val deserializerDecorator = new TypeRefAwareArangoDeserializerDecor(mockArangoDeserializer)
+
+    deserializerDecorator.deserialize(dummyVPack, dummyType)
+
+    verify(mockArangoDeserializer).deserialize(dummyVPack, dummyType)
+    verifyNoMoreInteractions(mockArangoDeserializer)
+  }
+
+  behavior of "deserialize(VPackSlice, TypeReference) method"
+
+  it should "delegate string types to the underlying deserializer" in {
+    val dummyVPack = mock[VPackSlice]
+    val stringTypeRef = new TypeReference[String] {}
+    val mockArangoDeserializer = mock[ArangoDeserializer]
+    val deserializerDecorator = new TypeRefAwareArangoDeserializerDecor(mockArangoDeserializer)
+
+    deserializerDecorator.deserialize(dummyVPack)(stringTypeRef)
+
+    verify(mockArangoDeserializer).deserialize(dummyVPack, classOf[String])
+    verifyNoMoreInteractions(mockArangoDeserializer)
+  }
+
+  it should "deserialize generic types" in {
+    val arangoJack = new ArangoJack {
+      configure(mapper => mapper.registerModule(new DefaultScalaModule))
+    }
+    val deserializerDecorator = new TypeRefAwareArangoDeserializerDecor(arangoJack)
+
+    val result = deserializerDecorator.deserialize[GenericContainer[Blah]](vpackParser.fromJson(
+      """
+        |{
+        |  "offset": 42,
+        |  "totalCount": 777,
+        |  "item": {"a": 0, "b": "bbb"},
+        |  "items": [
+        |    {"a": 111, "b": "foo"},
+        |    {"a": 222, "b": "bar"}
+        |  ]
+        |}
+        |""".stripMargin
+    ))
+
+    result shouldEqual GenericContainer(
+      item = Blah(0, "bbb"),
+      items = Seq(Blah(111, "foo"), Blah(222, "bar"))
+    )
+  }
+}
+
+object TypeRefAwareArangoDeserializerDecorSpec {
+  implicit val typeRef: TypeReference[GenericContainer[Blah]] = new TypeReference[GenericContainer[Blah]]() {}
+
+  case class GenericContainer[T: ClassTag](
+    items: Seq[T],
+    item: T
+  )
+
+  case class Blah(a: Int, b: String)
+}

producer-services/src/main/scala/za/co/absa/spline/producer/service/repo/ExecutionProducerRepositoryImpl.scala

@@ -17,6 +17,7 @@
 package za.co.absa.spline.producer.service.repo
 
 import com.arangodb.async.ArangoDatabaseAsync
+import com.fasterxml.jackson.core.`type`.TypeReference._
 import com.typesafe.scalalogging.LazyLogging
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Repository
@@ -42,7 +43,7 @@ class ExecutionProducerRepositoryImpl @Autowired()(
     val eventualPlanExists: Future[Boolean] =
       foxxRouter.get[Boolean](
         s"/spline/execution-plans/${executionPlan.id}/_exists", Map(
-          "discriminator" -> executionPlan.discriminator
+          "discriminator" -> executionPlan.discriminator.orNull
         ))
 
     val eventualPersistedDataSources: Future[Seq[DataSource]] = {
@@ -61,7 +62,7 @@ class ExecutionProducerRepositoryImpl @Autowired()(
         // No execution plan with the given ID found.
         // Let's insert one.
         val eppm = ExecutionPlanPersistentModelBuilder.toPersistentModel(executionPlan, persistedDSKeyByURI)
-        foxxRouter.post("/spline/execution-plans", eppm)
+        foxxRouter.post[Nothing]("/spline/execution-plans", eppm)
       }
     } yield ()
   })
@@ -73,7 +74,7 @@ class ExecutionProducerRepositoryImpl @Autowired()(
     val eventualEventExists: Future[Boolean] =
       foxxRouter.get[Boolean](
         s"/spline/execution-events/$eventKey/_exists", Map(
-          "discriminator" -> e.discriminator
+          "discriminator" -> e.discriminator.orNull
         ))
 
     for {
@@ -96,7 +97,7 @@ class ExecutionProducerRepositoryImpl @Autowired()(
           planKey = e.planId.toString,
           execPlanDetails = null // the value is populated below in the transaction script
         )
-        foxxRouter.post("/spline/execution-events", p)
+        foxxRouter.post[Nothing]("/spline/execution-events", p)
       }
     } yield ()
   })