Scala classOf generic type in Kafka json deserializer

Question

I'm writing a kafka json deserializer in scala using Jackson but am having a problem providing jackson's readValue() method the class of a generic type. For example:

...
import org.apache.kafka.common.serialization.Deserializer

class JsonDeserializer[T] extends Deserializer[Option[T]] {

  val mapper = (new ObjectMapper() with ScalaObjectMapper)
    .registerModule(DefaultScalaModule)
    .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
    .registerModule(new JavaTimeModule())
    .findAndRegisterModules()
    .asInstanceOf[ObjectMapper with ScalaObjectMapper]

  def deserialize(topic: String, bytes: Array[Byte]): Option[T] = {
    Option(bytes) match {
      case Some(b) => Some(mapper.readValue(bytes, classOf[T]))
      case None => None
    }

  }

  def configure(configs: java.util.Map[String, _], isKey: Boolean) {}
  def close(): Unit = {}
}

Notice the mapper.readValue(bytes, classOf[T]) in the deserialize method. Compilation fails with "class type required but T found".

How can this be done?

Answer 1

Generic types in Java are erased at runtime so there's no way to recover the Class without passing it in explicitly.

Well, explicitly as far as Java is concerned. You can use an (implicit ct: ClassTag[T]) or the shorthand [T: ClassTag] to (implicitly) obtain a ClassTag at construction time, which allows you to retrieve Class later on.

import scala.reflect._

class JsonDeserializer[T: ClassTag] extends Deserializer[Option[T]] {
  ...
    mapper.readValue(bytes, classTag[T].runtimeClass.asInstanceOf[Class[T]])