A few years ago, I was mentoring a developer who was tasked with building a real-time order processing system. Everything worked perfectly in staging, but the moment they hit production, the service entered a death spiral. A single malformed JSON message—what we call a "poison pill"—had entered the Kafka topic. The consumer would read the message, throw a `JsonParseException`, crash, restart, and immediately read the exact same message again. It happened every few seconds for three hours until we manually skipped the offset. It was a baptism by fire in the reality of distributed streaming: your code doesn't just need to handle the data; it needs to handle the failure of the data.

Bridging Java Clients with Scala Types

Now, here is the thing about Kafka: the gold standard is still the official Java client. While there are fantastic wrappers like FS2-Kafka or Alpakka, you need to understand the underlying Java mechanics to debug them. When you bring these into Scala, you're dealing with a lot of mutable Java collections and Properties objects. I always recommend wrapping the producer in a Scala class to hide that ugliness from the rest of your business logic.

To send a message, you'll use the KafkaProducer. The key is to remember that send() is asynchronous. It returns a Future (the Java version), which doesn't play nicely with Scala's scala.concurrent.Future. If you're just firing and forgetting, it's easy, but if you need a guarantee that the message hit the broker, you'll want to use a callback or call .get()—though be careful with the latter, as it blocks your thread.

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord, ProducerConfig}
import java.util.Properties

class OrderEventProducer(brokers: String) {
  val props = new Properties()
  props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
  props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
  props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")

  val producer = new KafkaProducer[String, String](props)

  def publishOrder(orderId: String, payload: String): Unit = {
    val record = new ProducerRecord("orders-topic", orderId, payload)
    producer.send(record, (metadata, exception) => {
      if (exception != null) println(s"Failed to send order $orderId: ${exception.getMessage}")
    })
  }
}

The Consumer Loop and Offset Management

Consuming is where things get tricky. Unlike a REST API where the request comes to you, in Kafka, you are the one asking for data. You have to implement a poll loop. This is where most developers stumble—they forget that the poll() method is the heartbeat of the consumer. If you spend too much time processing a single batch of messages without calling poll() again, Kafka will think your consumer has died and trigger a "rebalance," kicking you out of the group and assigning your partitions to someone else.

I usually suggest processing your records in a way that allows you to commit offsets explicitly. If you rely on enable.auto.commit = true, you might commit an offset for a message you haven't actually finished processing, leading to data loss if the app crashes. By disabling auto-commit, you tell Kafka, "I've actually handled this; now you can move the pointer forward."

Handling the Poison Pill

To avoid the crash loop I mentioned earlier, you must wrap your processing logic in a try-catch block inside the loop. If a message is unparseable, don't let the exception bubble up to the main loop. Instead, log the error, perhaps send the bad message to a "Dead Letter Queue" (DLQ) topic for manual inspection, and move on to the next record.

In Scala, we can make this elegant by treating the batch of records as a collection and using Either to represent success or failure. This keeps your pipeline clean and ensures that one bad piece of JSON doesn't bring down your entire infrastructure.

import org.apache.kafka.clients.consumer.{KafkaConsumer, ConsumerConfig, ConsumerRecords}
import java.time.Duration

val consumer = new KafkaConsumer[String, String](consumerProps)
consumer.subscribe(util.Arrays.asList("orders-topic"))

while (true) {
  val records: ConsumerRecords[String, String] = consumer.poll(Duration.ofMillis(100))
  
  records.forEach { record =>
    try {
      processOrder(record.value()) 
    } catch {
      case e: Exception => 
        println(s"Skipping poison pill at offset ${record.offset()}: ${e.getMessage}")
        // sendToDeadLetterQueue(record)
    }
  }
  consumer.commitSync()
}

📋 Practical Task

Build a Resilient Order Event Processor

Your task is to implement a Kafka consumer in Scala that can handle a stream of "Order" strings. Some of these strings will be valid JSON (e.g., {"id": "123", "amount": 50.0}), and some will be intentionally malformed (e.g., "INVALID_DATA").

Requirements:

• Create a OrderProcessor class that initializes a KafkaConsumer.
• Implement a run() method containing a while(true) loop that polls the orders-topic.
• Inside the loop, process each record. If the record is a valid JSON string, print "Processed order: [id]". If it is malformed, catch the exception, print "Error: Malformed message at offset [offset]", and ensure the loop continues to the next message without crashing.
• Disable ENABLE_AUTO_COMMIT_CONFIG and call commitSync() only after the entire batch of records from a single poll has been processed.
• Ensure you use a Duration of 100ms for the poll timeout.