Using Schema Registry for JSON Records

Schema Registry
1

Hi All,

I’m new to Confluent Cloud and deploying it using free trial of Confluent Kafka cluster.

I am trying to produce/consume the JSON messages from Kafka Topic using authenticated Schema Registry which can map Schema.

I am referring to the official documentation https://docs.confluent.io/platform/current/schema-registry/serdes-develop/serdes-json.html
but the JSON serializers mentioned here is not working.

I am getting below error:

Invalid value io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializer for configuration value.serializer: Class io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializer could not be found.

Could you please let us know if this has been deprecated or not valid anymore? Can someone please advise here?
Thanks.

2

See Class io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializer could not be found - Stack Overflow

3

@rmoff I have tried adding this dependency

libraryDependencies += “io.confluent” % “kafka-json-schema-serializer” % “6.2.0”

in my build.sbt file but it doesn’t get successful.

Extracting structure failed, reason: not ok build status: Error (BuildMessages(Vector(),Vector(BuildFailure(sbt task failed, see log for details)),Vector(),Vector(),Error))
sbt task failed, see log for details

scalaVersion := “2.12.5”
sbt.version = 1.5.2

Any suggestions please?

4

Hi @rmoff,

This just got resolved by adding

resolvers += “jitpack” at “https://jitpack.io”
libraryDependencies += “org.everit.json” % “org.everit.json.schema” % “1.5.1”

But now even with kafka-json-schema-serializer , the fields of the json records are not getting properly registered.

This is what I am trying to do:

case class User(id: Int,name: String,contact: Int, dept: String)
val usr: User = User(1,“abc”,123,“A”)

val producer = new KafkaProducer<String, User>(props)
val record= new ProducerRecord(topicName, “key”, usr)
producer.send(record,callback).get()

KEY_SERIALIZER_CLASS_CONFIG=>“org.apache.kafka.common.serialization.StringSerializer”
VALUE_SERIALIZER_CLASS_CONFIG=>“io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializer”)

But the values are coming as {} in my kafka topic. If I try to print the record part, I see this getting sent: key=key, value=User(1,“abc”,123,“A”)
Also, in the confluent schema registry the schema is coming as:

{
“additionalProperties”: false,
“properties”: {},
“title”: “User”,
“type”: “object”
}

Not sure why it is not correctly sending the records. Any suggestions?
I mainly want the schema registry to register all the fields of json records

{
“additionalProperties”: false,
“properties”: {“id”: { “type”: “integer”},…},
“title”: “User”,
“type”: “object”
}

Thanks.

5

You can try annotating the fields of the case class with @JsonProperty, like

case class User(@JsonProperty("id") id: Int, @JsonProperty("name") name: String, @JsonProperty("contact") contact: Int, @JsonProperty("dept") dept: String)

See java - Using Jackson to (De)-serialize a Scala Case Class - Stack Overflow