Building Robust Kafka Applications with Spring Boot, and Avro Schema Registry

As a software architect, designing solutions that are scalable, maintainable, and resilient is paramount. In the world of event-driven architectures, Apache Kafka has become a cornerstone for high-throughput, low-latency data streaming. However, simply sending raw bytes over Kafka topics can lead to data inconsistency and make future evolution a nightmare. This is where schema support comes into play.

In this comprehensive guide, we’ll walk through how to build a Spring Boot application that leverages Spring Kafka annotations, manages its dependencies with Gradle, and enforces a strong data contract using Apache Avro and the Confluent Schema Registry. By the end, you’ll have a clear understanding of how to create a robust and evolvable Kafka data pipeline.

Why Schema Support with Kafka?

Apache Kafka’s fundamental design is to treat messages as opaque byte arrays. While this provides incredible flexibility and performance, it offloads the responsibility of understanding message content to the producers and consumers. Without a defined schema:

• Data Inconsistency: Different producers might send data in slightly different formats, leading to consumers failing to parse messages.
• Schema Evolution Challenges: Changing message structures (adding/removing fields) becomes a nightmare, potentially breaking downstream applications.
• Lack of Type Safety: Developers work with raw bytes or generic maps, losing the benefits of compile-time type checking.
• No Centralized Contract: There’s no single source of truth for your data’s structure, hindering collaboration and data governance.

Apache Avro provides a compact binary serialization format with a rich schema definition language (using JSON). When coupled with a Schema Registry, it solves these problems by providing:

• Strong Data Contracts: Schemas enforce the structure of your data.
• Efficient Serialization: Avro’s binary format is highly efficient, reducing network bandwidth and storage.
• Schema Evolution: The Schema Registry manages schema versions and ensures compatibility rules (backward, forward, full compatibility) are met, allowing your data model to evolve gracefully.
• Type Safety: Avro-generated classes allow you to work with strongly typed objects in your Java code.

Prerequisites

Before we start coding, ensure you have the following running on your machine:

1. Java Development Kit (JDK 17+): Spring Boot 3.x requires JDK 17 or higher.
2. Gradle: Our build tool of choice.
3. Docker & Docker Compose: This is the easiest way to spin up a local Kafka cluster and Confluent Schema Registry.

Setting up Kafka and Schema Registry with Docker Compose

Create a docker-compose.yml file in your project root or a separate infra directory:

# docker-compose.yml
version: '3.7'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    hostname: kafka
    container_name: kafka
    ports:
      - "9092:9092" # Internal listener
      - "9093:9093" # External listener for applications
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_CONFLUENT_BALANCER_TOPIC_REPLICATION_FACTOR: 1
    depends_on:
      - zookeeper

  schema-registry:
    image: confluentinc/cp-schema-registry:7.5.0
    hostname: schema-registry
    container_name: schema-registry
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'kafka:9092'
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081
    depends_on:
      - kafka

To start these services, navigate to the directory containing docker-compose.yml in your terminal and run:

docker-compose up -d

This will start ZooKeeper, Kafka, and the Schema Registry in detached mode.

Project Setup with Gradle

First, initialize a new Spring Boot project (e.g., using Spring Initializr or your IDE).

build.gradle Configuration

Here’s the essential build.gradle content, including the Avro plugin for code generation:

// build.gradle
plugins {
    id 'java'
    id 'org.springframework.boot' version '3.3.1' // Use a recent Spring Boot version
    id 'io.spring.dependency-management' version '1.1.5'
    id 'com.github.davidmc24.gradle.plugin.avro' version '1.8.0' // Avro plugin
}

group = 'com.example'
version = '0.0.1-SNAPSHOT'
sourceCompatibility = '17'

repositories {
    mavenCentral()
    maven {
        url 'https://packages.confluent.io/maven/' // Confluent Maven repository for serializers
    }
}

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-web'
    implementation 'org.springframework.kafka:spring-kafka'
    implementation 'io.confluent:kafka-avro-serializer:7.5.0' // Confluent Avro serializer
    implementation 'org.apache.avro:avro:1.11.1' // Apache Avro library

    testImplementation 'org.springframework.boot:spring-boot-starter-test'
    testImplementation 'org.springframework.kafka:spring-kafka-test'
    testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
}

// Avro plugin configuration
avro {
    fieldVisibility = 'PRIVATE' // Generate private fields with public getters/setters
    outputDir = file("$buildDir/generated/source/avro") // Output directory for generated Java classes
}

sourceSets {
    main {
        java {
            srcDirs += files("$buildDir/generated/source/avro") // Add generated source to compilation path
        }
    }
}

tasks.named('test') {
    useJUnitPlatform()
}

Key parts of the Gradle setup:

• com.github.davidmc24.gradle.plugin.avro: This plugin is crucial. It automatically generates Java classes from your .avsc (Avro Schema) files.
• avro {} block: Configures the Avro plugin. We set fieldVisibility for cleaner generated code and specify outputDir to a build-generated folder.
• sourceSets {} block: We add the outputDir to the main Java source set. This tells Gradle to include the generated Avro Java classes in your project’s compilation path, making them available for use in your Spring application.
• Confluent Maven Repository: Necessary to resolve the kafka-avro-serializer dependency.

After configuring build.gradle, run gradle clean build or gradle generateAvro (if you explicitly use the Avro plugin’s task) to ensure the setup is correct and the plugin is recognized.

Define Your Avro Schema (.avsc)

Create a directory src/main/resources/avro and define your Avro schema file (e.g., PlayerCharacter.avsc). This schema will serve as the contract for messages exchanged over Kafka.

// src/main/resources/avro/PlayerCharacter.avsc
{
  "type": "record",
  "name": "PlayerCharacter",
  "namespace": "com.example.dnd.model",
  "doc": "Represents a player character in an AD&D 5E world expansion.",
  "fields": [
    {
      "name": "id",
      "type": "string",
      "doc": "Unique identifier for the character."
    },
    {
      "name": "name",
      "type": "string",
      "doc": "The character's name."
    },
    {
      "name": "race",
      "type": "string",
      "doc": "The character's race (e.g., Elf, Dwarf, Human)."
    },
    {
      "name": "characterClass",
      "type": "string",
      "doc": "The character's class (e.g., Fighter, Wizard, Rogue)."
    },
    {
      "name": "level",
      "type": "int",
      "default": 1,
      "doc": "The character's current level."
    },
    {
      "name": "hitPoints",
      "type": "int",
      "doc": "The character's current hit points."
    },
    {
      "name": "alignment",
      "type": "string",
      "default": "Neutral",
      "doc": "The character's moral and ethical alignment."
    }
  ]
}

After placing this file, run gradle build. The Avro plugin will generate PlayerCharacter.java in build/generated/source/avro/com/example/dnd/model/.

Spring Boot Configuration (application.yml)

With Kafka now configured as beans in AppConfig.java, your application.yml becomes much leaner regarding Kafka specifics. You’ll only need to define properties that are not managed directly by your bean definitions or are global Spring Boot properties.

# src/main/resources/application.yml
# No Kafka producer/consumer specific properties here,
# as they are now configured as @Bean definitions in AppConfig.java
# Any other general Spring Boot properties can remain here.
server:
  port: 8080 # Example: if you want to explicitly set the server port

Kafka Configuration as Beans (AppConfig.java)

To manage Kafka configuration programmatically as Spring beans, create an AppConfig.java class. This provides more flexibility and centralized control over your Kafka client properties.

// src/main/java/com/example/dnd/config/AppConfig.java
package com.example.dnd.config;

import com.example.dnd.model.PlayerCharacter;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.*;

import java.util.HashMap;
import java.util.Map;

@Configuration
@EnableKafka // Enables Spring's Kafka listener infrastructure
public class AppConfig {

    // Kafka broker address
    private static final String BOOTSTRAP_SERVERS = "localhost:9093";
    // Schema Registry URL
    private static final String SCHEMA_REGISTRY_URL = "http://localhost:8081";
    // Consumer group ID
    private static final String CONSUMER_GROUP_ID = "dnd-character-group";

    /**
     * Configures the Kafka ProducerFactory for sending PlayerCharacter messages.
     * This factory defines how producers are created, including serializers and Schema Registry integration.
     *
     * @return DefaultKafkaProducerFactory for String key and PlayerCharacter value.
     */
    @Bean
    public ProducerFactory<String, PlayerCharacter> producerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
        // Schema Registry configuration for the producer
        props.put("schema.registry.url", SCHEMA_REGISTRY_URL);
        // Optional: auto.register.schemas defaults to true, but can be explicitly set
        // props.put("auto.register.schemas", true);
        return new DefaultKafkaProducerFactory<>(props);
    }

    /**
     * Configures the KafkaTemplate, which is the high-level API for sending messages to Kafka.
     * It uses the producerFactory defined above.
     *
     * @param producerFactory The ProducerFactory bean.
     * @return KafkaTemplate for sending messages.
     */
    @Bean
    public KafkaTemplate<String, PlayerCharacter> kafkaTemplate(ProducerFactory<String, PlayerCharacter> producerFactory) {
        return new KafkaTemplate<>(producerFactory);
    }

    /**
     * Configures the Kafka ConsumerFactory for receiving PlayerCharacter messages.
     * This factory defines how consumers are created, including deserializers and Schema Registry integration.
     *
     * @return DefaultKafkaConsumerFactory for String key and PlayerCharacter value.
     */
    @Bean
    public ConsumerFactory<String, PlayerCharacter> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_ID);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // Start from the beginning if no offset
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true); // Auto commit offsets for simplicity
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
        // Schema Registry configuration for the consumer
        props.put("schema.registry.url", SCHEMA_REGISTRY_URL);
        // Crucial: tells Avro deserializer to return specific Avro-generated classes
        props.put("specific.avro.reader", true);
        return new DefaultKafkaConsumerFactory<>(props);
    }

    /**
     * Configures the ConcurrentKafkaListenerContainerFactory, which is used by @KafkaListener annotations.
     * It uses the consumerFactory defined above.
     *
     * @param consumerFactory The ConsumerFactory bean.
     * @return ConcurrentKafkaListenerContainerFactory for message listening.
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, PlayerCharacter> kafkaListenerContainerFactory(
            ConsumerFactory<String, PlayerCharacter> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, PlayerCharacter> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        // Optional: Configure batch listening if needed (requires List<PlayerCharacter> in listener method)
        // factory.setBatchListener(true);
        return factory;
    }
}

Spring Kafka Producer (with KafkaTemplate)

This class remains unchanged. The KafkaTemplate will now be automatically wired by Spring from the AppConfig bean definition.

// src/main/java/com/example/dnd/producer/PlayerCharacterProducer.java
package com.example.dnd.producer;

import com.example.dnd.model.PlayerCharacter; // This is the generated Avro class
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class PlayerCharacterProducer {

    private static final String TOPIC_NAME = "dnd-characters";

    // Spring automatically injects KafkaTemplate based on your AppConfig bean definition
    private final KafkaTemplate<String, PlayerCharacter> kafkaTemplate;

    public PlayerCharacterProducer(KafkaTemplate<String, PlayerCharacter> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    /**
     * Sends a PlayerCharacter object to the Kafka topic.
     * The KafkaAvroSerializer will handle serialization and schema registration.
     *
     * @param key The message key (e.g., character ID).
     * @param character The PlayerCharacter object to send.
     */
    public void sendPlayerCharacter(String key, PlayerCharacter character) {
        System.out.println(String.format("Producing message to topic %s: key=%s, value=%s", TOPIC_NAME, key, character));
        kafkaTemplate.send(TOPIC_NAME, key, character);
        // For production, you'd typically add a callback to handle successful sends or failures:
        /*
        kafkaTemplate.send(TOPIC_NAME, key, character).whenComplete((result, ex) -> {
            if (ex == null) {
                System.out.println("Sent message successfully: topic=" + result.getRecordMetadata().topic() +
                                   ", partition=" + result.getRecordMetadata().partition() +
                                   ", offset=" + result.getRecordMetadata().offset());
            } else {
                System.err.println("Failed to send message: " + ex.getMessage());
            }
        });
        */
    }
}

Spring Kafka Consumer (with @KafkaListener Annotations)

This class also remains unchanged. The @KafkaListener annotation will automatically use the kafkaListenerContainerFactory bean defined in AppConfig.

// src/main/java/com/example/dnd/consumer/PlayerCharacterConsumer.java
package com.example.dnd.consumer;

import com.example.dnd.model.PlayerCharacter; // This is the generated Avro class
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;

@Service
public class PlayerCharacterConsumer {

    /**
     * Listens for messages on the 'dnd-characters' topic.
     * The KafkaAvroDeserializer automatically deserializes the message bytes
     * into a PlayerCharacter object based on the schema from the Schema Registry.
     *
     * @param character The deserialized PlayerCharacter object (payload).
     * @param topic The topic from which the message was received.
     * @param partition The partition from which the message was received.
     * @param offset The offset of the message within the partition.
     */
    @KafkaListener(topics = "dnd-characters", groupId = "dnd-character-group")
    public void listen(@Payload PlayerCharacter character,
                       @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                       @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
                       @Header(KafkaHeaders.OFFSET) long offset) {

        System.out.println(String.format("### CONSUMED MESSAGE ###"));
        System.out.println(String.format("  Topic: %s, Partition: %d, Offset: %d", topic, partition, offset));
        System.out.println(String.format("  Character ID: %s", character.getId()));
        System.out.println(String.format("  Name: %s", character.getName()));
        System.out.println(String.format("  Race: %s", character.getRace()));
        System.out.println(String.format("  Class: %s", character.getCharacterClass()));
        System.out.println(String.format("  Level: %d", character.getLevel()));
        System.out.println(String.format("  HP: %d", character.getHitPoints()));
        System.out.println(String.format("  Alignment: %s", character.getAlignment()));
        System.out.println("########################\n");
    }

    // You can define multiple listeners for different topics or groups,
    // or even for different message types if your configuration allows it.

    // Example of a batch listener (requires additional configuration in KafkaConfig)
    /*
    import java.util.List;
    @KafkaListener(topics = "dnd-characters-batch", groupId = "dnd-batch-group", containerFactory = "batchKafkaListenerContainerFactory")
    public void listenBatch(List<PlayerCharacter> characters) {
        System.out.println(String.format("Consumed %d characters in batch.", characters.size()));
        characters.forEach(character -> System.out.println("- Batch: " + character.getName()));
    }
    */
}

Example Usage (REST Controller)

This class also remains unchanged, as it depends on PlayerCharacterProducer, which is now a Spring-managed bean via AppConfig.

// src/main/java/com/example/dnd/controller/CharacterController.java
package com.example.dnd.controller;

import com.example.dnd.model.PlayerCharacter; // The generated Avro class
import com.example.dnd.producer.PlayerCharacterProducer;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.util.UUID;

@RestController
@RequestMapping("/api/characters")
public class CharacterController {

    private final PlayerCharacterProducer producer;

    // Spring automatically injects PlayerCharacterProducer
    public CharacterController(PlayerCharacterProducer producer) {
        this.producer = producer;
    }

    /**
     * REST endpoint to create and send a new PlayerCharacter to Kafka.
     * @param character The PlayerCharacter object sent in the request body.
     * @return A response indicating success and the character ID.
     */
    @PostMapping
    public ResponseEntity<String> createPlayerCharacter(@RequestBody PlayerCharacter character) {
        // Assign a unique ID for the character if not provided
        if (character.getId() == null || character.getId().isEmpty()) {
            character.setId(UUID.randomUUID().toString());
        }

        // Send the character to Kafka
        producer.sendPlayerCharacter(character.getId(), character);

        return new ResponseEntity<>("Player Character sent to Kafka with ID: " + character.getId(), HttpStatus.CREATED);
    }
}

Main Spring Boot Application

The main application class remains the same, as @EnableKafka is still needed, and it’s on AppConfig now.

// src/main/java/com/example/dnd/SpringKafkaAvroDemoApplication.java
package com.example.dnd;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
// @EnableKafka is now on AppConfig.java
// import org.springframework.kafka.annotation.EnableKafka;

@SpringBootApplication
public class SpringKafkaAvroDemoApplication {

    public static void main(String[] args) {
        SpringApplication.run(SpringKafkaAvroDemoApplication.class, args);
    }
}

Running the Application

1. Start Docker Containers:docker-compose up -d
2. Build your Spring Boot application:./gradlew clean build
This step is vital as it triggers the Avro plugin to generate your PlayerCharacter.java class.
3. Run the Spring Boot application:java -jar build/libs/spring-kafka-avro-demo-0.0.1-SNAPSHOT.jar
(Adjust the JAR name if your version differs)You should see Spring Boot starting up and your Kafka listeners initialized.
4. Send a message using cURL or Postman:Open another terminal and send a POST request to your /api/characters endpoint.
You should see output in your Spring Boot application’s console indicating that the message was produced and then consumed, with the PlayerCharacter object correctly deserialized and its fields accessed.

curl -X POST http://localhost:8080/api/characters \ -H "Content-Type: application/json" \ -d '{ "name": "Arion", "race": "Half-Elf", "characterClass": "Bard", "level": 3, "hitPoints": 25, "alignment": "Chaotic Good" }'

Observing Schema Registration

You can verify the schema was registered by checking the Schema Registry’s REST API:

curl http://localhost:8081/subjects

You should see ["dnd-characters-value"] listed.

Then, to see the registered schema:

curl http://localhost:8081/subjects/dnd-characters-value/versions/1

This will show you the JSON schema that was registered.

Benefits of This Approach

• Centralized Configuration: All Kafka-related client properties (bootstrap servers, serializers, deserializers, Schema Registry URL, group ID) are defined in one dedicated @Configuration class. This improves organization, especially in larger applications.
• Programmatic Control: You have more fine-grained control over bean creation and can inject other services or dynamic values if needed, which is harder with application.yml.
• Testability: Kafka beans can be easily mocked or replaced with test doubles for unit and integration testing.
• Type Safety: Working directly with PlayerCharacter objects eliminates runtime ClassCastException issues and provides compile-time checks.
• Data Consistency: The Schema Registry ensures that all data produced to the dnd-characters topic adheres to the defined Avro schema.
• Schema Evolution: If you need to add a new optional field (e.g., background) to your PlayerCharacter.avsc, you can update the schema, regenerate the Java class, and new producers will use the new schema. Older consumers will still be able to read the messages, and new consumers will correctly interpret both old and new messages thanks to Avro’s compatibility rules and the Schema Registry.
• Performance: Avro’s binary serialization is compact and efficient, ideal for high-volume data streams.
• Clear Contracts: The .avsc file acts as a centralized, human-readable, and machine-enforceable contract for your data, improving collaboration across teams.
• Spring Boot Convenience: Spring Boot and Spring Kafka annotations dramatically simplify the configuration and development of Kafka producers and consumers.

By integrating Avro and the Confluent Schema Registry into your Spring Kafka applications with Gradle, you build a robust, future-proof, and easily manageable data streaming solution for your world-expansion data or any other complex domain.

Understanding Kafka Consumer Groups and Their Relation to Messages

In Kafka, consumer groups are a fundamental concept that enables scalable and fault-tolerant consumption of messages. They dictate how messages published to topics are distributed among multiple consumer instances.

What is a Consumer Group ID?

A Consumer Group ID is a unique identifier that you assign to a set of consumer instances that collectively consume messages from one or more Kafka topics. In our AppConfig.java, we’ve defined it as dnd-character-group.

private static final String CONSUMER_GROUP_ID = "dnd-character-group";

And then used it in our ConsumerConfig:

props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_ID);

How Consumer Groups Relate to Messages and Topics

1. Shared Consumption: All consumers within the same consumer group share the responsibility of reading messages from a topic. When a message is published to a topic, only one consumer instance within a given consumer group will receive that message.
2. Scalability:
   • Within a Group: If you have a topic with multiple partitions (e.g., 3 partitions), and you run multiple consumer instances within the same group (e.g., 3 instances), each instance can consume from a different partition. This allows for parallel processing and increased throughput. If you add more consumer instances than partitions, the extra instances will be idle.
   • Across Groups: If you need different applications or logical services to process the same set of messages from a topic independently, each application should belong to a different consumer group. Each group will receive its own copy of all messages.
   Consider our dnd-characters topic:
   • If you run two instances of our Spring Boot application, both configured with group-id: dnd-character-group, Kafka will distribute the messages between them. Each message will be processed by only one of these instances.
   • If you deploy another entirely different application (e.g., a “Character Archiver” service) that also needs to process all PlayerCharacter messages, you would configure it with a different group-id (e.g., character-archiver-group). Both dnd-character-group and character-archiver-group would receive a complete copy of all messages published to dnd-characters.
3. Offset Management: Kafka keeps track of the “offset” (the last consumed message position) for each consumer group per partition. This ensures that:
   • If a consumer instance fails, another instance in the same group can take over its partitions and resume consumption from the last committed offset, preventing data loss or reprocessing.
   • Messages are not missed or duplicated within a group during rebalances (when consumers join or leave a group).
4. Load Balancing: Kafka automatically handles the distribution of partitions among the active consumers in a group. When a consumer joins or leaves a group, or when a partition is added to a topic, Kafka triggers a “rebalance” to redistribute the partitions evenly among the remaining or new consumers.

In summary, the consumer-group-id is vital for defining the consumption behavior and scalability patterns of your Kafka consumers. It allows you to build highly available and horizontally scalable applications that reliably process event streams.

Sending Messages to Multiple Consumer Groups

This is a common point of confusion for those new to Kafka! You do not explicitly “send a message to multiple consumer groups.” Instead, you send a single message to a Kafka topic, and then each consumer group that is subscribed to that topic will receive its own copy of that message.

Let’s illustrate with our dnd-characters topic:

1. Producer’s Role: Our PlayerCharacterProducer sends a PlayerCharacter message to the dnd-characters topic:producer.sendPlayerCharacter(character.getId(), character);
The producer’s job is simply to publish the message to the designated topic. It has no knowledge of how many consumer groups exist or are subscribed to that topic.
2. Kafka’s Role in Distribution: Once the message is in the dnd-characters topic, Kafka’s core functionality takes over:
   • Independent Copies: For each unique consumer group that is subscribed to dnd-characters, Kafka will deliver a copy of that message.
   • Group-Internal Distribution: Within each consumer group, Kafka ensures that only one consumer instance receives that particular message, distributing messages across the group’s instances (and their assigned partitions) for parallel processing.

Practical Example:

Imagine you have our Spring Boot application (with group-id: dnd-character-group) consuming PlayerCharacter events. Now, let’s say you introduce two new applications:

• Application A (Character Archiver): This application is responsible for archiving all character data to a long-term storage solution. It subscribes to the dnd-characters topic with group-id: character-archiver-group.
• Application B (Game Analytics): This application analyzes character creation patterns for game balancing purposes. It subscribes to the dnd-characters topic with group-id: game-analytics-group.

When our PlayerCharacterProducer sends a single PlayerCharacter message to dnd-characters:

• One instance of the Spring Boot application (within dnd-character-group) will receive and process it.
• One instance of the Character Archiver application (within character-archiver-group) will receive and process it.
• One instance of the Game Analytics application (within game-analytics-group) will receive and process it.

All three applications, representing three distinct consumer groups, will independently process the same message from the topic. The producer simply publishes once, and Kafka handles the fan-out to all subscribed groups.

This decoupling is a key strength of Kafka, allowing different services to react to the same events without direct dependencies between them. You achieve “sending to multiple groups” by simply configuring multiple distinct consumer groups to subscribe to the same topic.