AugmentedMessageDeserializer

AugmentedMessageDeserializer.scala

class AugmentedMessageDeserializer extends MessageKafkaDeserializer[GenericRecord] with LazyLogging {

  override def deserialize(record: ConsumerRecord[Array[Byte], Array[Byte]]): GenericRecord = {
    val genericRecord: Try[GenericRecord] = retrieveGenericRecord(record)

    genericRecord match {
      case Success(value) => value
      case Failure(exception) =>
        val key = new String(record.key(), UTF_8)
        val value = new String(record.value(), UTF_8)

        logger.error(
          s"operation='deserialize', message='Error in kafka record', key=$key, value=$value",
          exception
        )

        null
    }
  }

  override def getProducedType: TypeInformation[GenericRecord] = TypeExtractor.getForClass(classOf[GenericRecord])

  private def retrieveGenericRecord(record: ConsumerRecord[Array[Byte], Array[Byte]]): Try[GenericRecord] = Try {
    val rawContent = new String(record.value(), UTF_8)
    val rawJson: Json = parse(rawContent).getOrElse(throw new IllegalArgumentException(s"Invalid json on $rawContent"))

    val timestamp = rawJson.hcursor
      .downField("event")
      .get[String](TIMESTAMP.name)
      .toOption
      .getOrElse(INVALID.name)

    timestamp match {
      case INVALID.name => throw new NoSuchElementException(s"Not found timestamp on $rawContent")
      case _ =>
        val genericRecord = new GenericData.Record(GenericRecordSchema.schema)
        genericRecord.put(KEY.name, new String(record.key(), UTF_8))
        genericRecord.put(MESSAGE.name, record.value())
        genericRecord.put(HEADERS.name, createJsonStringOfHeaders(record.headers()))
        genericRecord.put(TIMESTAMP.name, timestamp)

        genericRecord
    }
  }

  private def createJsonStringOfHeaders(headers: Headers): String =
    headers.toArray
      .map(
        header => header.key() -> header.value()
      )
      .toMap
      .asJson
      .toString
}