scalaapache-kafkacase-classspray-jsonalpakka

Scala adding an element to JsValue before convertT[class]


I have a Kafka consumed record that will be parsed as JsValue with spray.json in scala, but I also have some data in the record's header, and I want to do:

  1. Consume record with Alpakka Kafka library (done)

  2. parse as json of type JsValue: kafkaRecord.record.value().parseJson (also done)

  3. Append the record's header to that JsValue (HERE IS THE MAIN CONCERN)

    Header = kafkaRecord.record.headers()

    Appending should include key[String]: value(header)

  4. convert to pre-defined case class using [JsValue].convertTo[<case class>]

Here is the consumed record for example:

{"id": 23, "features": "features_23"}

and want to append to it the Header to be as:

{"id": 23, "features": "features_23", "header_data":"Header_23"}

Then convert to case class:

case class recordOfKafka(id: Int, features: String, header_data: String)

Solution

  • Assuming some functions that get the string value of the record and header:

    @ def getKafkaRecord(): String = """
      {"id":1,"features":"feature_23"}
      """ 
    defined function getKafkaRecord
    
    @ def getKafkaHeader(): String = """
      {"header_data":"header_23"}
      """ 
    defined function getKafkaHeader
    

    And the model with formatter:

    @ import spray.json._
    @ import DefaultJsonProtocol._
    
    @ case class SomeModel(id: Int, features: String, header_data: String) 
    defined class SomeModel
    
    @ implicit val someModelFormatter = jsonFormat3(SomeModel) 
    someModelFormatter: RootJsonFormat[SomeModel] = spray.json.ProductFormatsInstances$$anon$3@6c1e40d9
    

    You can parse record value and header in separate json objects, and then combine their fields together:

    @ // parse as json object, convert to map
    @ JsObject(
        getKafkaRecord.parseJson.asJsObject.fields ++
          getKafkaHeader.parseJson.asJsObject.fields
      ).convertTo[SomeModel] 
    res15: SomeModel = SomeModel(
      id = 1,
      features = "feature_23",
      header_data = "header_23"
    )
    
    

    One thing to take care about is that, if you want to use .asJsObject you must be sure that the records are json objects, like if you receive:

    [{"key":"value"},{"key":"value_2"}]
    

    It'll throw exceptions since this is an json array and cannot be converted to json object! You can also use try or pattern matching for conversion to JsObject to be safer.