If you've spent any time in production with Kafka, you know the "poison pill" scenario: someone changes a field from an Integer to a String in the producer, and suddenly every single consumer in your pipeline starts crashing with a serialization error. It's a nightmare to clean up. That's why we use a Schema Registry. It acts as a central source of truth, ensuring that producers and consumers are speaking the same language without having to hardcode schemas into every single microservice.

For this lesson, we're going to build a simple event producer and consumer for a UserActivity stream. We'll use Apache Avro, as it's the gold standard for this workflow.

Defining our UserActivity schema

First, we don't write Scala classes for our data; we write an Avro schema file. This allows the Registry to version the schema. I'm creating a file called user_activity.avsc. I want to track the user's ID, the action they took, and a timestamp.

{
  "type": "record",
  "name": "UserActivity",
  "namespace": "com.example.events",
  "fields": [
    {"name": "userId", "type": "string"},
    {"name": "action", "type": "string"},
    {"name": "timestamp", "type": "long"}
  ]
}

Once this is defined, I use a plugin like sbt-avrohugger to generate the Scala case classes from this JSON. This gives us a type-safe UserActivity object to work with in our code, while the Registry handles the actual bytes on the wire.

Configuring the Producer for Avro

Now, let's get the producer running. The key difference here is that we can't use the standard StringSerializer or ByteArraySerializer. We need the KafkaAvroSerializer provided by Confluent.

val props = new Properties()
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getName)
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[KafkaAvroSerializer].getName)
props.put("schema.registry.url", "http://localhost:8081")

val producer = new KafkaProducer[String, UserActivity](props)
val activity = UserActivity.newBuilder()
  .setUserId("user-123")
  .setAction("click_signup")
  .setTimestamp(System.currentTimeMillis())
  .build()

producer.send(new ProducerRecord("user-activities", "user-123", activity))

Correcting my serialization slip-up

I'll be honest: when I first wrote this snippet a few minutes ago, I forgot to add the schema.registry.url property. I hit run and immediately got a KafkaException: Error constructing producer because the KafkaAvroSerializer had no idea where to register the schema. It doesn't default to a local port; it needs an explicit URL.

If you see an error mentioning that the serializer cannot find the registry configuration, don't waste time debugging your Kafka brokers. Check your Properties object. The serializer needs that URL to check if the schema already exists or if it needs to register a new version.

Wiring up the Consumer to respect the schema

On the consumer side, we do the opposite. We use the KafkaAvroDeserializer. One detail I always emphasize is the specific.avro.reader property. If you set this to false, Kafka returns a GenericRecord (which feels like a Map). If you set it to true, it maps the data back into our generated UserActivity Scala class.

val consumerProps = new Properties()
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "activity-monitor")
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, classOf[StringDeserializer].getName)
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, classOf[KafkaAvroDeserializer].getName)
consumerProps.put("schema.registry.url", "http://localhost:8081")
consumerProps.put("specific.avro.reader", "true")

val consumer = new KafkaConsumer[String, UserActivity](consumerProps)
consumer.subscribe(java.util.Collections.singletonList("user-activities"))

while (true) {
  val records = consumer.poll(java.time.Duration.ofMillis(100))
  records.forEach { record =>
    println(s"User ${record.value().getUserId} performed ${record.value().getAction}")
  }
}

By using the Registry, the consumer doesn't need the .avsc file bundled in its source code. It just asks the Registry: "I see schema ID 5 in this message header; can you give me the definition for that?" and the Registry handles the rest.

📋 Practical Task

Exercise: Implementing Schema Evolution for UserActivity

In a real-world scenario, requirements change. Your task is to evolve the UserActivity schema without breaking the existing consumer.

• Modify the Schema: Update user_activity.avsc to add a new field called deviceType (a string). To ensure backward compatibility, you must provide a default value of "unknown".
• Update the Producer: Update your Scala producer to populate this new deviceType field (e.g., "mobile" or "desktop").
• Verify Compatibility: Run the original consumer (the one without the deviceType field in its generated classes) and verify that it can still read the new messages produced by the updated producer without crashing.
• Update the Consumer: Finally, regenerate the Scala classes for the consumer and update the print statement to include the deviceType.