60-Day Kafka 4 Learning Plan · Week 1 Capstone · End-to-end lab
Sources: Kafka: The Definitive Guide Ch.2–4 · kafka.apache.org/43 · Spring Kafka Docs
Goal
Build a complete order processing pipeline using Java 21, Spring Boot 3, and Apache Kafka 4 running in KRaft mode — no ZooKeeper, no legacy coordination overhead. Kafka UI gives you a browser-based view into the topic while the application runs.
Kafka 4 and KRaft Mode
Kafka 4.0 removes ZooKeeper entirely. KRaft (Kafka Raft) replaces it with an internal consensus protocol where Kafka manages its own metadata without an external coordinator.
Why this matters for developers:
- One less service to run and maintain
- Faster startup and leader election
- Simpler Docker Compose files (one container instead of two)
- Better operational visibility — everything is in Kafka
In KRaft mode a node can be a broker, a controller, or both. For development, combining both roles in a single node is standard.
What We’re Building
POST /api/v1/orders
│
▼
OrderController HTTP layer — validates input, generates orderId
│
▼
OrderProducer Publishes OrderEvent to "orders" topic
│ Key = orderId (ensures partition affinity per order)
▼
┌───────────────────────────────────────────────┐
│ orders topic (3 partitions) │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │Partition0│ │Partition1│ │Partition2│ │
│ └──────────┘ └──────────┘ └──────────┘ │
└───────────────────────────────────────────────┘
│
▼
OrderConsumer @KafkaListener — logs orderId, product, partition, offsetMessage key = orderId. Using a key routes all events for the same order to the same partition, preserving per-order ordering guarantees across retries and downstream consumers.
Step 1 — Infrastructure: Kafka 4 + Kafka UI
services:
kafka:
image: apache/kafka:4.0.0
hostname: kafka
container_name: kafka
ports:
- "9092:9092"
environment:
# KRaft — no ZooKeeper
KAFKA_NODE_ID: 1
KAFKA_PROCESS_ROLES: broker,controller
CLUSTER_ID: 4L6g3nShT-eMCtK--X86sw
# Two listeners:
# INTERNAL — Docker network (used by Kafka UI)
# EXTERNAL — host machine (used by the Spring Boot app)
KAFKA_LISTENERS: CONTROLLER://kafka:9093,INTERNAL://kafka:29092,EXTERNAL://0.0.0.0:9092
KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:29092,EXTERNAL://localhost:9092
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
healthcheck:
test: ["CMD-SHELL", "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 > /dev/null 2>&1"]
interval: 15s
timeout: 10s
retries: 5
start_period: 30s
kafka-ui:
image: provectuslabs/kafka-ui:latest
container_name: kafka-ui
ports:
- "8090:8080"
depends_on:
kafka:
condition: service_healthy
environment:
KAFKA_CLUSTERS_0_NAME: local-kafka-4
KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:29092Two-listener pattern explained:
Kafka needs to advertise different addresses depending on who is connecting:
The INTERNAL listener handles broker-to-broker traffic and container-to-container connections. The EXTERNAL listener handles connections from the host machine. The CONTROLLER listener handles KRaft consensus traffic.
CLUSTER_ID is a base64-encoded UUID that identifies the Kafka cluster. Generate your own with kafka-storage.sh random-uuid or use any stable string for development.
Start the stack:
docker compose up -dOpen Kafka UI at http://localhost:8090 after a few seconds. You’ll see an empty local-kafka-4 cluster with no topics yet.
Step 2 — Maven Dependencies
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>3.5.0</version>
</parent>
<properties>
<java.version>21</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-validation</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>The spring-kafka version is managed by the Spring Boot BOM — no explicit version needed.
Step 3 — Application Configuration
spring:
application:
name: kafka-lab
kafka:
bootstrap-servers: localhost:9092
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
acks: all
retries: 3
properties:
delivery.timeout.ms: 120000
linger.ms: 5
batch.size: 16384
consumer:
group-id: order-consumer-group
auto-offset-reset: earliest
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
properties:
spring.json.trusted.packages: "com.boottechsolutions.kafkalab.event"
spring.json.value.default.type: com.boottechsolutions.kafkalab.event.OrderEvent
listener:
ack-mode: record
concurrency: 3Configuration decisions worth explaining:
acks: all — the producer waits for all in-sync replicas to acknowledge before considering a send successful. In a single-node dev cluster this is equivalent to acks: 1, but it’s the right default for production code.
linger.ms: 5 — the producer waits 5ms before sending, allowing small batches to accumulate. Reduces network round trips at low cost to latency.
auto-offset-reset: earliest — consumers that join without a committed offset start from the beginning. This is appropriate for development but should be latest for production consumers joining after go-live.
spring.json.trusted.packages — Spring Kafka’s JsonDeserializer rejects type headers from untrusted packages by default. Restrict this to your event packages rather than using *.
listener.concurrency: 3 — one listener thread per partition. Since the orders topic has 3 partitions, this saturates available parallelism. Setting concurrency higher than the partition count wastes thread resources.
Step 4 — Project Structure
src/main/java/com/boottechsolutions/kafkalab/
├── KafkaLabApplication.java
├── config/
│ └── KafkaTopicConfig.java # creates "orders" topic on startup
├── controller/
│ └── OrderController.java # POST /api/v1/orders
├── consumer/
│ └── OrderConsumer.java # @KafkaListener
├── domain/
│ └── OrderStatus.java # enum
├── dto/
│ ├── CreateOrderRequest.java
│ ├── CreateOrderResponse.java
│ └── ErrorResponse.java
├── event/
│ └── OrderEvent.java # Kafka message payload
├── exception/
│ └── GlobalExceptionHandler.java
└── producer/
└── OrderProducer.java # KafkaTemplate wrapperStep 5 — Topic Configuration
Spring Boot’s Kafka auto-configuration picks up NewTopic beans and creates the topic on startup (idempotent — ignored if the topic already exists):
@Configuration
public class KafkaTopicConfig {
public static final String ORDERS_TOPIC = "orders";
@Bean
public NewTopic ordersTopic() {
return TopicBuilder.name(ORDERS_TOPIC)
.partitions(3)
.replicas(1)
.build();
}
}3 partitions allow up to 3 concurrent consumers in the order-consumer-group. Adding more partitions later requires a rebalance — plan partition count with expected consumer parallelism in mind.
replicas(1) is correct for a single-node dev cluster. In production with a 3-node cluster you’d set this to 3.
Step 6 — The Order Event
The Kafka message payload is a Java 21 record serialized as JSON:
public record OrderEvent(
String orderId,
String customerId,
String product,
int quantity,
BigDecimal amount,
OrderStatus status,
Instant createdAt
) {}Records are ideal for Kafka events: immutable, no boilerplate, naturally serializable to JSON. The JsonSerializer/JsonDeserializer handles the Jackson mapping automatically.
Step 7 — The Producer
@Component
@RequiredArgsConstructor
@Slf4j
public class OrderProducer {
private final KafkaTemplate<String, OrderEvent> kafkaTemplate;
public void send(OrderEvent event) {
kafkaTemplate.send(KafkaTopicConfig.ORDERS_TOPIC, event.orderId(), event)
.thenAccept(result -> log.info(
"Published orderId={} partition={} offset={}",
event.orderId(),
result.getRecordMetadata().partition(),
result.getRecordMetadata().offset()
))
.exceptionally(ex -> {
log.error("Failed to publish orderId={}: {}", event.orderId(), ex.getMessage());
return null;
});
}
}kafkaTemplate.send(topic, key, value) returns a CompletableFuture<SendResult>. The .thenAccept callback logs the assigned partition and offset after the broker acknowledges receipt. The .exceptionally handler logs failures without propagating them to the caller.
Key = orderId. Kafka hashes the key to select the partition: partition = hash(key) % numPartitions. All events for the same order go to the same partition in the same order. This matters when a downstream system needs to process all events for an order sequentially.
Step 8 — The Consumer
@Component
@Slf4j
public class OrderConsumer {
@KafkaListener(
topics = "${spring.kafka.topic.orders:orders}",
groupId = "${spring.kafka.consumer.group-id}",
containerFactory = "kafkaListenerContainerFactory"
)
public void consume(
@Payload OrderEvent event,
@Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
@Header(KafkaHeaders.OFFSET) long offset,
@Header(KafkaHeaders.RECEIVED_TOPIC) String topic
) {
log.info(
"Received orderId={} product={} qty={} amount={} status={} | topic={} partition={} offset={}",
event.orderId(),
event.product(),
event.quantity(),
event.amount(),
event.status(),
topic,
partition,
offset
);
}
}@Payload extracts the deserialized OrderEvent from the Kafka record value. The @Header annotations extract Kafka metadata injected by the framework — partition number, offset, and topic name. Including these in logs is essential for tracing which record a consumer processed.
The groupId references the YAML property so it stays consistent with the consumer configuration. If you have multiple consumer groups (e.g., an analytics consumer and an order-fulfillment consumer), each would use its own groupId and receive all messages independently.
Step 9 — The Controller
@RestController
@RequestMapping("/api/v1/orders")
@RequiredArgsConstructor
@Slf4j
public class OrderController {
private final OrderProducer producer;
@PostMapping
public ResponseEntity<CreateOrderResponse> create(@Valid @RequestBody CreateOrderRequest request) {
String orderId = UUID.randomUUID().toString();
OrderEvent event = new OrderEvent(
orderId,
request.customerId(),
request.product(),
request.quantity(),
request.amount(),
OrderStatus.PENDING,
Instant.now()
);
producer.send(event);
log.info("Order accepted orderId={} customerId={}", orderId, request.customerId());
return ResponseEntity
.status(HttpStatus.ACCEPTED)
.body(new CreateOrderResponse(orderId, OrderStatus.PENDING, "Order accepted for processing"));
}
}202 Accepted, not 201 Created. The order doesn’t exist yet in any persistent store — it’s been accepted for asynchronous processing. Returning 201 would imply the resource was created, which it wasn’t. 202 communicates “we received it and will process it.”
The request DTO with validation:
public record CreateOrderRequest(
@NotBlank(message = "customerId must not be blank")
String customerId,
@NotBlank(message = "product must not be blank")
@Size(max = 200, message = "product name must not exceed 200 characters")
String product,
@Min(value = 1, message = "quantity must be at least 1")
int quantity,
@NotNull(message = "amount must not be null")
@DecimalMin(value = "0.01", message = "amount must be greater than zero")
BigDecimal amount
) {}Running the Lab
# 1. Start Kafka 4 + Kafka UI
docker compose up -d
# 2. Start the application
./mvnw spring-boot:runExpected API response (202):
Producer log:
INFO 19866 --- [kafka-lab] [-lab-producer-1] c.b.kafkalab.producer.OrderProducer : Published orderId=c361a53d-c892-4281-9df2-6ec0b85570f2 partition=1 offset=0Consumer log:
INFO 19866 --- [kafka-lab] [ntainer#0-1-C-1] c.b.kafkalab.consumer.OrderConsumer : Received orderId=c361a53d-c892-4281-9df2-6ec0b85570f2 product=Wireless Keyboard qty=2 amount=79.99 status=PENDING | topic=orders partition=1 offset=0Note the partition number is identical in both logs — same key, same partition.Kafka UI — browse the topic:
Navigate to http://localhost:8090 → local-kafka-4 → Topics → orders. You’ll see the message in the partition it was assigned to, with the full JSON payload and all headers.
Validation error:
Submit multiple orders and watch partition distribution:
for i in {1..9}; do
curl -s -X POST http://localhost:8080/api/v1/orders \
-H "Content-Type: application/json" \
-d "{\"customerId\": \"cust-00$i\", \"product\": \"Item $i\", \"quantity\": $i, \"amount\": 9.99}" > /dev/null
doneCheck Kafka UI — orders distribute across all 3 partitions based on hash(orderId) % 3.
Performance Considerations
Partition Count and Consumer Parallelism
Partition count is a ceiling on consumer parallelism within a group. 3 partitions → 3 consumers can process simultaneously. Partitions cannot be reduced after creation without deleting the topic. Plan high — 10–20 partitions for topics expected to carry significant load.
Batching and Throughput
linger.ms: 5 and batch.size: 16384 (16 KB) improve producer throughput by sending records in batches. Increasing linger.ms to 20–50ms and batch.size to 65536+ significantly improves throughput at the cost of slightly higher latency.
Consumer Lag
Consumer lag (the gap between the latest offset and the consumer’s committed offset) is the primary health metric for Kafka consumers. Monitor it with:
docker exec kafka /opt/kafka/bin/kafka-consumer-groups.sh \
--bootstrap-server localhost:9092 \
--describe \
--group order-consumer-groupA growing lag indicates the consumer can’t keep up. Options: increase listener.concurrency, add more partitions, or scale consumer instances.
Idempotent Producer
With acks: all and retries: 3, the producer may occasionally send a message more than once on transient failures. Enable idempotence to prevent duplicates:
spring:
kafka:
producer:
properties:
enable.idempotence: trueThis guarantees exactly-once delivery to the broker. Idempotence requires acks: all and max.in.flight.requests.per.connection: 5 or less — both are already set correctly.
Common Pitfalls
auto-offset-reset: latest in development. With latest, a consumer that starts after messages were published sees nothing. Use earliest during development so the consumer replays all messages. Set latest in production where consumers should only process new messages.
trusted.packages: "*" in production. Trusting all packages for JSON deserialization is a security risk. Always restrict to your specific event packages.
Topic does not exist on first startup. If the Spring Boot app starts before Kafka is healthy, the NewTopic bean creation attempt fails. The Docker Compose depends_on with condition: service_healthy prevents this. Without it, add retry logic or start Kafka first.
Consumer group rebalance on deployment. When a new consumer instance joins (blue-green deployment), Kafka rebalances partition assignments across the consumer group. During rebalance, consumption pauses briefly. Use cooperative-sticky rebalancing to minimize the pause:
spring:
kafka:
consumer:
properties:
partition.assignment.strategy: org.apache.kafka.clients.consumer.CooperativeStickyAssignorLosing messages on consumer exception. With ack-mode: record, an unhandled exception causes the record to be reprocessed. Wrap consumer logic in try-catch and log errors rather than propagating. For dead-letter queue patterns, configure a DeadLetterPublishingRecoverer.
Missing @DirtiesContext in integration tests. Without it, EmbeddedKafka state bleeds between test classes, causing flaky consumer offset behavior. Always add @DirtiesContext to @EmbeddedKafka tests.
Key Takeaways
- Kafka 4 in KRaft mode removes ZooKeeper — Docker Compose goes from two infrastructure services to one
- The two-listener pattern (
INTERNALfor containers,EXTERNALfor the host) is the correct networking model for Docker-based development - Use
orderIdas the message key to guarantee per-order partition affinity and sequential processing; matchlistener.concurrencyto the partition count to maximize parallelism
Week 1 Review
The complete source code is available on GitHub.
Support me through GitHub Sponsors.
Next
➡️ Week 2: Kafka Internals – Day 8 — KRaft Metadata Log
