From 3a682e496ed38b1d5601bd642086332d7fb8fa66 Mon Sep 17 00:00:00 2001 From: Ankur Mhatre Date: Sat, 29 Aug 2026 09:47:25 +0530 Subject: [PATCH] Add the kafka-basics module --- .gitignore | 2 + LICENSE | 21 +++ README.md | 30 +++++ kafka-basics/README.md | 60 +++++++++ kafka-basics/docs/01-the-on-ramp.md | 50 +++++++ kafka-basics/docs/02-producing.md | 67 +++++++++ kafka-basics/docs/03-serialisation.md | 82 +++++++++++ kafka-basics/docs/04-keys-and-partitions.md | 51 +++++++ kafka-basics/docs/05-consuming.md | 59 ++++++++ kafka-basics/docs/06-acknowledgement.md | 79 +++++++++++ kafka-basics/docs/07-testing.md | 84 ++++++++++++ kafka-basics/docs/output/effective-config.txt | 71 ++++++++++ kafka-basics/docs/output/key-to-partition.txt | 7 + .../docs/output/serialised-payload.txt | 1 + kafka-basics/docs/output/tests.txt | 4 + kafka-basics/pom.xml | 84 ++++++++++++ kafka-basics/scripts/run-all.sh | 15 +++ .../kafkabasics/KafkaBasicsApplication.java | 20 +++ .../com/ankurm/kafkabasics/OrderConsumer.java | 48 +++++++ .../com/ankurm/kafkabasics/OrderEvent.java | 22 +++ .../com/ankurm/kafkabasics/OrderProducer.java | 66 +++++++++ .../src/main/resources/application.yaml | 33 +++++ .../kafkabasics/EffectiveConfigTest.java | 127 ++++++++++++++++++ .../kafkabasics/KeysAndPartitionsTest.java | 95 +++++++++++++ .../com/ankurm/kafkabasics/ManualAckTest.java | 85 ++++++++++++ .../ankurm/kafkabasics/SerialisationTest.java | 91 +++++++++++++ .../TestcontainersConfiguration.java | 45 +++++++ .../src/test/resources/application-test.yaml | 4 + 28 files changed, 1403 insertions(+) create mode 100644 .gitignore create mode 100644 LICENSE create mode 100644 README.md create mode 100644 kafka-basics/README.md create mode 100644 kafka-basics/docs/01-the-on-ramp.md create mode 100644 kafka-basics/docs/02-producing.md create mode 100644 kafka-basics/docs/03-serialisation.md create mode 100644 kafka-basics/docs/04-keys-and-partitions.md create mode 100644 kafka-basics/docs/05-consuming.md create mode 100644 kafka-basics/docs/06-acknowledgement.md create mode 100644 kafka-basics/docs/07-testing.md create mode 100644 kafka-basics/docs/output/effective-config.txt create mode 100644 kafka-basics/docs/output/key-to-partition.txt create mode 100644 kafka-basics/docs/output/serialised-payload.txt create mode 100644 kafka-basics/docs/output/tests.txt create mode 100644 kafka-basics/pom.xml create mode 100755 kafka-basics/scripts/run-all.sh create mode 100644 kafka-basics/src/main/java/com/ankurm/kafkabasics/KafkaBasicsApplication.java create mode 100644 kafka-basics/src/main/java/com/ankurm/kafkabasics/OrderConsumer.java create mode 100644 kafka-basics/src/main/java/com/ankurm/kafkabasics/OrderEvent.java create mode 100644 kafka-basics/src/main/java/com/ankurm/kafkabasics/OrderProducer.java create mode 100644 kafka-basics/src/main/resources/application.yaml create mode 100644 kafka-basics/src/test/java/com/ankurm/kafkabasics/EffectiveConfigTest.java create mode 100644 kafka-basics/src/test/java/com/ankurm/kafkabasics/KeysAndPartitionsTest.java create mode 100644 kafka-basics/src/test/java/com/ankurm/kafkabasics/ManualAckTest.java create mode 100644 kafka-basics/src/test/java/com/ankurm/kafkabasics/SerialisationTest.java create mode 100644 kafka-basics/src/test/java/com/ankurm/kafkabasics/TestcontainersConfiguration.java create mode 100644 kafka-basics/src/test/resources/application-test.yaml diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..b1c13f4 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +target/ +*.log diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..aa5473f --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Ankur Mhatre + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..e001608 --- /dev/null +++ b/README.md @@ -0,0 +1,30 @@ +# spring-messaging-demo + +Companion code for the messaging series on [ankurm.com](https://ankurm.com). Each directory is a +self-contained Maven project for one article, with its own `pom.xml`, its own numbered +documentation chapters, and its own captured output under `docs/output/` — regenerated by that +module's `scripts/run-all.sh`, never typed by hand. + +| Module | Article | What it demonstrates | +|---|---|---| +| [`kafka-basics/`](kafka-basics/README.md) | [Spring Boot 4.1 and Apache Kafka: Producer, Consumer and Serialisation from Scratch](https://ankurm.com/spring-boot-4-1-kafka-producer-consumer-serialisation/) | The on-ramp: what the starter gives you, the two Jackson serializer families, where a key lands and why, and which defaults are Kafka's rather than Spring's | + +They are meant to be read in order. `kafka-basics` establishes that the default acknowledgement +mode is `BATCH` and therefore that delivery is at-least-once; everything the error-handling +module does exists because of that one sentence. + +## Common ground + +All modules target the same verified stack: **JDK 25** (Temurin 25.0.4.1+1), **Spring Boot +4.1.1**, **Spring Framework 7.0.9**. Versions were read from `maven-metadata.xml` on Maven +Central and from Boot's own `spring-boot-dependencies` POM, rather than from release +announcements. + +Every module runs its broker without Docker, so the transcripts can be regenerated on any +machine with a JDK: Kafka via the in-process KRaft broker in `spring-kafka-test`, RabbitMQ via a +real broker started by the module's own scripts. Each module also ships a Testcontainers +configuration for the cases where the deployed image is what you need to test against. + +## Licence + +MIT — see [LICENSE](LICENSE). diff --git a/kafka-basics/README.md b/kafka-basics/README.md new file mode 100644 index 0000000..30d901a --- /dev/null +++ b/kafka-basics/README.md @@ -0,0 +1,60 @@ +# `kafka-basics` — producer, consumer and serialisation, from nothing + +Companion project for +[**Spring Boot 4.1 and Apache Kafka: Producer, Consumer and Serialisation from Scratch**](https://ankurm.com/spring-boot-4-1-kafka-producer-consumer-serialisation/) +on ankurm.com. + +Eight tests against a **real Kafka broker**, started in-process in KRaft mode. No Docker daemon, +no local Kafka install, no ZooKeeper. `./scripts/run-all.sh` regenerates everything under +[`docs/output/`](docs/output/). + +## Versions + +| | Version | Notes | +|---|---|---| +| JDK | 25 (Temurin 25.0.4.1+1) | current LTS | +| Spring Boot | 4.1.1 | GA of the 4.1 line was 10 June 2026 | +| Spring Kafka | 4.1.1 | Boot-managed | +| kafka-clients | **4.2.1** | Boot-managed. Central has 4.3.1; let the BOM decide | +| Jackson | 3.1.5 (`tools.jackson`) | why `JacksonJsonSerializer` and not `JsonSerializer` | +| Testcontainers | 2.0.5 | artifact is `testcontainers-kafka`, not `kafka` | + +Versions were read from `repo1.maven.org/.../maven-metadata.xml` and from Boot's own +`spring-boot-dependencies` POM, not from release announcements. + +## Quickstart + +```bash +./scripts/run-all.sh # runs every test and regenerates docs/output/ +mvn test # the same thing without the capture +``` + +## Documentation + +1. [The on-ramp, and the dependency that is not the one you remember](docs/01-the-on-ramp.md) +2. [Producing, and the return value everybody throws away](docs/02-producing.md) +3. [Serialisation: two Jackson families](docs/03-serialisation.md) +4. [Keys and partitions: the ordering guarantee in disguise](docs/04-keys-and-partitions.md) +5. [Consuming](docs/05-consuming.md) +6. [Acknowledgement, and a property that is not where you look for it](docs/06-acknowledgement.md) +7. [Testing without installing Kafka](docs/07-testing.md) + +## Captured output + +| File | Produced by | +|---|---| +| [`effective-config.txt`](docs/output/effective-config.txt) | client defaults vs Boot's overrides vs effective | +| [`key-to-partition.txt`](docs/output/key-to-partition.txt) | real placement against a real broker | +| [`serialised-payload.txt`](docs/output/serialised-payload.txt) | the Jackson 3 wire format | +| [`tests.txt`](docs/output/tests.txt) | 8 tests | + +## Four things this module exists to prove + +1. **`spring-kafka` alone gives you no auto-configuration under Boot 4.** You need + `spring-boot-starter-kafka`; the symptom is a missing `KafkaTemplate` bean. +2. **`JsonSerializer` is the Jackson 2 one.** It cannot write a `java.time.Instant` with its + default mapper. `JacksonJsonSerializer` is the Jackson 3 one and can. +3. **The default partitioner is `murmur2 & 0x7fffffff`, not `Math.abs(murmur2)`.** They disagree + on two of the six keys in the committed transcript. +4. **`ConsumerFactory.isAutoCommit()` answers `true` while no consumer auto-commits.** The + container overrides the property per-consumer and never tells the factory. diff --git a/kafka-basics/docs/01-the-on-ramp.md b/kafka-basics/docs/01-the-on-ramp.md new file mode 100644 index 0000000..fb9507b --- /dev/null +++ b/kafka-basics/docs/01-the-on-ramp.md @@ -0,0 +1,50 @@ +[Module README](../README.md) · [Producing →](02-producing.md) + +# 1. The on-ramp, and the dependency that is not the one you remember + +Under Spring Boot 3 you added `org.springframework.kafka:spring-kafka` and got a +`KafkaTemplate`. Under Boot 4 you get this: + +``` +No qualifying bean of type 'org.springframework.kafka.core.KafkaTemplate' available +``` + +Boot 4 split the auto-configurations out of `spring-boot-autoconfigure` into per-technology +modules. Kafka's now lives in `spring-boot-kafka`, package +`org.springframework.boot.kafka.autoconfigure`, and a bare `spring-kafka` dependency does not +bring it. The application compiles, the context starts, and there is simply no template. + +```xml + + org.springframework.boot + spring-boot-starter-kafka + +``` + +That is the fix, and the same shape applies elsewhere: `spring-boot-starter-amqp` for RabbitMQ, +`spring-boot-starter-restclient` for `RestClient.Builder`. The rule of thumb for Boot 4 is that +if you are depending on a library directly rather than through a Boot starter, you are probably +missing its auto-configuration. + +## What the starter actually gives you + +| Bean | Comes from | Notes | +|---|---|---| +| `KafkaTemplate` | `KafkaAutoConfiguration` | typed by your injection point | +| `ProducerFactory` / `ConsumerFactory` | `KafkaAutoConfiguration` | built from `spring.kafka.*` | +| `KafkaListenerContainerFactory` | `KafkaAnnotationDrivenConfiguration` | what `@KafkaListener` binds to | +| `KafkaAdmin` | `KafkaAutoConfiguration` | creates `NewTopic` beans at startup | + +`KafkaAdmin` is worth knowing about early: declare a `NewTopic` bean and Boot creates the topic +on startup with the partition count and replication factor you asked for. It will **not** change +an existing topic's partition count, so a `NewTopic` bean that disagrees with the cluster is +silently ignored rather than applied. + +## Versions + +Boot 4.1.1 manages Spring Kafka **4.1.1** and kafka-clients **4.2.1**. Note the second one: +Maven Central has kafka-clients 4.3.1, and overriding the managed version to reach it is the +kind of change that works until it does not. Let the BOM decide. + +[Producing →](02-producing.md) diff --git a/kafka-basics/docs/02-producing.md b/kafka-basics/docs/02-producing.md new file mode 100644 index 0000000..993e1df --- /dev/null +++ b/kafka-basics/docs/02-producing.md @@ -0,0 +1,67 @@ +[← The on-ramp](01-the-on-ramp.md) · [Module README](../README.md) · [Serialisation →](03-serialisation.md) + +# 2. Producing, and the return value everybody throws away + +[`OrderProducer`](../src/main/java/com/ankurm/kafkabasics/OrderProducer.java) has three send +methods because there are exactly three useful answers to "when do I find out this failed". + +```java +public void sendAndForget(OrderEvent event) { + this.template.send(TOPIC, event.orderId(), event); // returns a CompletableFuture. Ignored. +} +``` + +`KafkaTemplate.send` is asynchronous and returns a `CompletableFuture>`. +Discarding it discards the only notification you will get. The method returns normally, the +record may never reach the broker, and nothing in your logs says so — the producer's own retry +and expiry messages are at `WARN` under `org.apache.kafka`, which most applications turn down. + +This is the most common way to lose messages in a Spring Kafka application, and it looks like +correct code. + +```java +this.template.send(TOPIC, event.orderId(), event) + .whenComplete((result, ex) -> { /* log, meter, compensate */ }); +``` + +Handle the future, or block on it when the caller genuinely must not proceed without a durable +write: + +```java +SendResult result = this.template.send(record).get(); +result.getRecordMetadata().partition(); // where it landed +result.getRecordMetadata().offset(); // and at what offset +``` + +Blocking costs a network round trip plus the replication acknowledgement, so it belongs at the +edges of a system, not inside a loop. + +## Durability comes from defaults you did not set + +Spring Boot sets **nothing** on the producer beyond bootstrap servers and serializers. Here is +the effective configuration, printed by +[`EffectiveConfigTest`](../src/test/java/com/ankurm/kafkabasics/EffectiveConfigTest.java) and +committed at [`docs/output/effective-config.txt`](output/effective-config.txt): + +``` +property kafka-clients default set by Spring Boot +acks all - +enable.idempotence true - +retries 2147483647 - +max.in.flight.requests.per.connection 5 - +delivery.timeout.ms 120000 - +``` + +Since Kafka 3.0 the client defaults are `acks=all` and `enable.idempotence=true`, so a stock +Boot application already has a durable, deduplicating producer. Two consequences: + +- **You do not need to set `acks=all`.** It is already on. +- **An old runbook that sets `acks=1` or `retries=0` is now a downgrade.** Those lines were + written when the defaults were weaker, and deleting them makes the system safer, which is the + opposite of how configuration usually ages. + +`delivery.timeout.ms` at two minutes is the one worth revisiting: it is the total budget for a +send including retries, and a request that exceeds it fails permanently. If your future never +completes, that is the clock you are waiting on. + +[Serialisation →](03-serialisation.md) diff --git a/kafka-basics/docs/03-serialisation.md b/kafka-basics/docs/03-serialisation.md new file mode 100644 index 0000000..c40b452 --- /dev/null +++ b/kafka-basics/docs/03-serialisation.md @@ -0,0 +1,82 @@ +[← Producing](02-producing.md) · [Module README](../README.md) · [Keys and partitions →](04-keys-and-partitions.md) + +# 3. Serialisation: two Jackson families, and the one your IDE will suggest is the wrong one + +Kafka moves `byte[]`. Everything else is a `Serializer` and a `Deserializer`, and under Spring +Boot 4 there is a fork in the road that no tutorial mentions yet. + +Spring Kafka 4.1 ships **two complete JSON families**: + +| Class | Jackson | Mapper type | +|---|---|---| +| `JsonSerializer` / `JsonDeserializer` / `JsonSerde` | 2.x | `com.fasterxml.jackson.databind.ObjectMapper` | +| `JacksonJsonSerializer` / `JacksonJsonDeserializer` / `JacksonJsonSerde` | 3.x | `tools.jackson.databind.json.JsonMapper` | + +Spring Boot 4 is a Jackson 3 application. `spring-boot-starter-jackson` brings +`tools.jackson.core:jackson-databind` 3.1.5. But the class named `JsonSerializer` — the one +every existing example configures, and the one autocomplete offers first — is the Jackson 2 one. + +It does not fail at startup. It fails on the first payload containing a `java.time` value: + +``` +org.apache.kafka.common.errors.SerializationException: Can't serialize data + [OrderEvent[orderId=o-1, ..., placedAt=2026-08-29T10:15:30Z]] for topic [orders] +Caused by: com.fasterxml.jackson.databind.exc.InvalidDefinitionException: + Java 8 date/time type `java.time.Instant` not supported by default: + add Module "com.fasterxml.jackson.datatype:jackson-datatype-jsr310" to enable handling +``` + +The message tells you to add a Jackson 2 module, and doing so works — and leaves you running two +Jackson stacks, one for your HTTP layer and one for your messaging layer, with independent +configuration. The better fix is one word: + +```yaml +spring: + kafka: + producer: + value-serializer: org.springframework.kafka.support.serializer.JacksonJsonSerializer + consumer: + value-deserializer: org.springframework.kafka.support.serializer.JacksonJsonDeserializer +``` + +Jackson 3 handles `java.time` with no module and no configuration, and `BigDecimal` keeps its +scale through a round trip — both asserted in +[`SerialisationTest`](../src/test/java/com/ankurm/kafkabasics/SerialisationTest.java): + +``` +{"orderId":"o-1","customerId":"c-1","amount":10.00,"placedAt":"2026-08-29T10:15:30Z"} +``` + +`amount` is `10.00`, not `10.0` and not `"10.00"`. Use `BigDecimal` for money and this survives; +use `double` and it does not. + +## The type header + +The three-argument `serialize(topic, headers, data)` overload writes a `__TypeId__` header +naming the class: + +```java +headers.lastHeader("__TypeId__") -> "com.ankurm.kafkabasics.OrderEvent" +``` + +The consumer reads it and builds that class. Which is convenient and is also a remote-code- +selection primitive, so the deserializer refuses any class outside its trusted packages: + +``` +... is not in the trusted packages: [java.util, java.lang] +``` + +Three ways out, in descending order of preference: + +1. `spring.json.trusted.packages: com.ankurm.kafkabasics` — an allow-list of your own packages. +2. `spring.json.value.default.type` — ignore the header, always build this class. Best when + producer and consumer are owned by different teams, because it makes the consumer's contract + its own decision rather than the producer's. +3. `spring.json.trusted.packages: "*"` — do not. + +Option 2 has a second benefit worth stating plainly: the `__TypeId__` header couples the two +services by **fully qualified class name**. Renaming or moving a class in the producer breaks +every consumer that trusts the header, at runtime, with no compile-time warning. `type.mapping` +(a logical name to class map on both sides) is the version of this that survives a refactor. + +[Keys and partitions →](04-keys-and-partitions.md) diff --git a/kafka-basics/docs/04-keys-and-partitions.md b/kafka-basics/docs/04-keys-and-partitions.md new file mode 100644 index 0000000..67e209c --- /dev/null +++ b/kafka-basics/docs/04-keys-and-partitions.md @@ -0,0 +1,51 @@ +[← Serialisation](03-serialisation.md) · [Module README](../README.md) · [Consuming →](05-consuming.md) + +# 4. Keys and partitions: the ordering guarantee in disguise + +Kafka orders records **within a partition**. Not within a topic. So the key is not a label — it +decides which records are ordered with respect to each other, and it is the most consequential +line in a producer. + +Same key, same partition, forever: + +``` +key partition murmur2 & 0x7fffffff % 3 Math.abs(murmur2) % 3 +o-1 0 0 0 +o-2 0 0 2 +o-3 1 1 1 +o-4 1 1 1 +o-5 2 2 0 +o-6 1 1 1 +``` + +From [`docs/output/key-to-partition.txt`](output/key-to-partition.txt), produced against a real +broker by +[`KeysAndPartitionsTest`](../src/test/java/com/ankurm/kafkabasics/KeysAndPartitionsTest.java). + +The default partitioner is `murmur2` of the **serialized key bytes**, masked positive, modulo +the partition count. Note the third and fourth columns: `& 0x7fffffff` and `Math.abs` disagree +on two of six keys, because clearing the sign bit is not the same number as negating it. If you +reimplement the partitioner to predict placement — for a test, for a migration, for a routing +table — `Math.abs` gives you the right answer about half the time, which is the worst available +failure mode. + +## Three consequences + +**A null key is not a key.** Records without one are spread across partitions by the sticky +partitioner, so nothing about their relative order is guaranteed. If two events describe the same +entity and you did not key them, they can be processed out of order by different consumers. + +**Adding partitions repartitions every key.** The modulus changes, so `o-2` moves. Anything +relying on per-key ordering loses it across the resize for records still in flight. Pick the +partition count with room to grow; changing it later is a data-ordering event, not a capacity +knob. + +**Key cardinality is your parallelism ceiling.** Keying by `customerId` when one customer is 40% +of traffic gives you a hot partition that no amount of consumer scaling fixes, because one +partition is consumed by exactly one member of a group. + +Choosing the key is choosing what must stay ordered. Order events by `orderId` if operations on +one order must not overtake each other; by `customerId` if that is true across a customer's +orders. Those are different systems. + +[Consuming →](05-consuming.md) diff --git a/kafka-basics/docs/05-consuming.md b/kafka-basics/docs/05-consuming.md new file mode 100644 index 0000000..f9c696e --- /dev/null +++ b/kafka-basics/docs/05-consuming.md @@ -0,0 +1,59 @@ +[← Keys and partitions](04-keys-and-partitions.md) · [Module README](../README.md) · [Acknowledgement →](06-acknowledgement.md) + +# 5. Consuming + +```java +@KafkaListener(topics = "orders", groupId = "orders-basic") +public void onOrder(@Payload OrderEvent event, + @Header(KafkaHeaders.RECEIVED_PARTITION) int partition, + @Header(KafkaHeaders.OFFSET) long offset) { ... } +``` + +There is no poll loop, no offset commit and no rebalance listener, because the container owns +all three. That is the actual value Spring Kafka adds over the plain client, and it is worth +knowing what it is doing on your behalf — see [chapter 6](06-acknowledgement.md). + +## The single most common "my listener never fires" + +`auto.offset.reset` decides where a consumer group with **no committed offset** starts. The +kafka-clients default is `latest`: skip everything already in the topic. So a brand-new group +attached to a topic full of records reads nothing, and the listener you just wrote never runs. + +```yaml +spring: + kafka: + consumer: + auto-offset-reset: earliest +``` + +Boot does not change this default; the value in this module's `application.yaml` does. It only +applies when there is no committed offset — once the group has committed once, this setting is +irrelevant, which is why the symptom disappears the first time it works and never comes back. + +The other reliable cause is a `groupId` mismatch between the annotation and +`spring.kafka.consumer.group-id`. The annotation wins, and a typo there creates a brand-new group +that then hits the paragraph above. + +## Concurrency and what it can and cannot buy + +`spring.kafka.listener.concurrency` creates that many consumers in the group, each on its own +thread. The ceiling is the partition count: with 3 partitions, `concurrency: 10` gives you three +working consumers and seven idle ones. There is no configuration that makes two threads consume +one partition, because that would break the ordering guarantee from +[chapter 4](04-keys-and-partitions.md). + +## Headers worth knowing + +| Constant | What it carries | +|---|---| +| `KafkaHeaders.RECEIVED_KEY` | the producer's key, through the key deserializer | +| `KafkaHeaders.RECEIVED_PARTITION` | which partition this came from | +| `KafkaHeaders.OFFSET` | the offset within that partition | +| `KafkaHeaders.RECEIVED_TIMESTAMP` | broker or producer timestamp, per topic config | +| `__TypeId__` | the class name the producer wrote — see [chapter 3](03-serialisation.md) | + +Declare the key as `required = false` unless you are certain every record has one. A `null` key +on a `required = true` header parameter fails the conversion, and the failure arrives as a +deserialization-time error rather than as a null. + +[Acknowledgement →](06-acknowledgement.md) diff --git a/kafka-basics/docs/06-acknowledgement.md b/kafka-basics/docs/06-acknowledgement.md new file mode 100644 index 0000000..0f192bf --- /dev/null +++ b/kafka-basics/docs/06-acknowledgement.md @@ -0,0 +1,79 @@ +[← Consuming](05-consuming.md) · [Module README](../README.md) · [Testing →](07-testing.md) + +# 6. Acknowledgement, and a property that is not where you look for it + +The container's default `AckMode` is **`BATCH`**: commit the offsets of the whole `poll()` batch +after the listener has returned for every record in it. Confirmed at runtime by +[`EffectiveConfigTest`](../src/test/java/com/ankurm/kafkabasics/EffectiveConfigTest.java): + +``` +=== listener container === +ackMode BATCH +groupId orders-basic +``` + +That gives at-least-once delivery. A crash after processing and before the commit redelivers the +batch, which is why your listener must be idempotent — and why +[the error-handling article](https://ankurm.com/spring-kafka-4-1-error-handling-dlt-retry-topics/) +is a necessary sequel rather than an optional one. + +## Where `enable.auto.commit` actually lives + +This one is worth the detour, because every debugging session about commits starts in the wrong +place. + +- The **kafka-clients default** for `enable.auto.commit` is `true`. +- **Spring Boot does not set it.** It is absent from `ConsumerFactory.getConfigurationProperties()`, + before the containers start and after. +- **`ConsumerFactory.isAutoCommit()` returns `true`** on a stock Boot 4.1 application, because it + reads that same absent key and falls back to the client default. +- And yet **no consumer auto-commits**, because `ListenerConsumer.determineAutoCommit` checks + whether the factory config contains the key and, when it does not, calls + `setProperty("enable.auto.commit", "false")` on the **per-container `Properties`** handed to + `createConsumer`. + +So the shared factory never learns, its public accessor answers the opposite of the truth, and +the real value lives in an override map you cannot reach from application code. All four of +those statements are asserted in the test, including the counter-intuitive one: + +```java +assertThat(this.consumerFactory.isAutoCommit()).isTrue(); +``` + +If you ever need to know whether a running consumer auto-commits, read the consumer's own +metrics or its startup log line, not the factory. + +## Manual acknowledgement + +```yaml +spring: + kafka: + listener: + ack-mode: MANUAL +``` + +```java +@KafkaListener(topics = "orders") +void onOrder(OrderEvent event, Acknowledgment acknowledgment) { + process(event); + acknowledgment.acknowledge(); +} +``` + +Nothing commits until `acknowledge()` runs, which is what you want when the work must be durable +before the offset moves. Two things to know: + +- Asking for `MANUAL` while auto-commit is genuinely enabled is an + `IllegalStateException` at container start. It does not happen on a stock configuration — + despite `isAutoCommit()` returning `true` — because `determineAutoCommit` sets the container + property to `false` before the check runs. [`ManualAckTest`](../src/test/java/com/ankurm/kafkabasics/ManualAckTest.java) + pins that down, because it is exactly the kind of interaction that would otherwise be a + surprise in production. +- **An `Acknowledgment` you forget to call stalls the partition.** Not immediately — the + container keeps polling until `max.poll.records` of un-acknowledged records accumulate. So the + symptom is a consumer that works for a while and then stops, which reads like a broker problem. + +`MANUAL_IMMEDIATE` commits synchronously on the consumer thread instead of at the end of the +batch. It is slower and it is the right choice when redelivery is genuinely expensive. + +[Testing →](07-testing.md) diff --git a/kafka-basics/docs/07-testing.md b/kafka-basics/docs/07-testing.md new file mode 100644 index 0000000..33a0556 --- /dev/null +++ b/kafka-basics/docs/07-testing.md @@ -0,0 +1,84 @@ +[← Acknowledgement](06-acknowledgement.md) · [Module README](../README.md) + +# 7. Testing without installing Kafka + +Two options, both real brokers, and the choice is less obvious than it looks. + +## `@EmbeddedKafka` — a real broker inside the JVM + +```java +@SpringBootTest +@EmbeddedKafka(topics = "orders", partitions = 3) +class KeysAndPartitionsTest { ... } +``` + +`spring-kafka-test` starts `EmbeddedKafkaKraftBroker` — the actual Apache Kafka broker classes, +in KRaft mode, in-process. No ZooKeeper, no container, no daemon. It binds a random port and +exposes it as `${spring.embedded.kafka.brokers}`: + +```yaml +spring: + kafka: + bootstrap-servers: ${spring.embedded.kafka.brokers} +``` + +It starts in about three seconds and needs nothing installed, which is why every transcript in +this module came from it and why `./scripts/run-all.sh` works on a machine with no Docker. + +## Testcontainers — the image you actually deploy + +```java +@TestConfiguration(proxyBeanMethods = false) +public class TestcontainersConfiguration { + + @Bean + @ServiceConnection + KafkaContainer kafkaContainer() { + return new KafkaContainer(DockerImageName.parse("apache/kafka:4.1.0")); + } + +} +``` + +`@ServiceConnection` registers the container's bootstrap servers as the application's, which +removes the `@DynamicPropertySource` block that older examples all carry. + +**Two coordinates changed recently and both will bite you:** + +```xml + + + org.testcontainers + testcontainers-kafka + test + +``` + +Testcontainers 2.x prefixed every module artifact with `testcontainers-`, and Boot 4.1.1 imports +`testcontainers-bom` 2.0.5, which manages only the new names. Using the old coordinate fails +with a Maven error that does not mention the rename: + +``` +'dependencies.dependency.version' for org.testcontainers:kafka:jar is missing +``` + +The class moved too: use `org.testcontainers.kafka.KafkaContainer` (Apache Kafka, KRaft), not +the older `org.testcontainers.containers.KafkaContainer` (Confluent images, ZooKeeper). + +## Which to use + +| | `@EmbeddedKafka` | Testcontainers | +|---|---|---| +| startup | ~3s | ~10s, plus image pull | +| needs Docker | no | yes | +| broker version | the client library's | whatever image you name | +| TLS, SASL, quotas, partitions | not modelled | real | + +Use `@EmbeddedKafka` for the bulk of a suite and Testcontainers for the handful of tests where +the difference between "the broker classes" and "the broker you deploy" matters. The +[`TestcontainersConfiguration`](../src/test/java/com/ankurm/kafkabasics/TestcontainersConfiguration.java) +in this module is compiled but not exercised by `run-all.sh`, because the machine that +regenerates `docs/output/` has no Docker daemon — which is itself the argument for keeping +both paths. + +[Module README](../README.md) diff --git a/kafka-basics/docs/output/effective-config.txt b/kafka-basics/docs/output/effective-config.txt new file mode 100644 index 0000000..0d87c34 --- /dev/null +++ b/kafka-basics/docs/output/effective-config.txt @@ -0,0 +1,71 @@ +=== producer === +property kafka-clients default set by Spring Boot +-------------------------------------------------------------------------------------------------------- +acks all - +enable.idempotence true - +retries 2147483647 - +max.in.flight.requests.per.connection 5 - +linger.ms 5 - +batch.size 16384 - +compression.type none - +delivery.timeout.ms 120000 - + +=== consumer === +property kafka-clients default set by Spring Boot +-------------------------------------------------------------------------------------------------------- +auto.offset.reset latest earliest +enable.auto.commit true - +max.poll.records 500 - +max.poll.interval.ms 300000 - +session.timeout.ms 45000 - +heartbeat.interval.ms 3000 - +isolation.level read_uncommitted read_uncommitted +partition.assignment.strategy [class org.apache.kafka.clients.consumer.RangeAssignor, class org.apache.kafka.clients.consumer.CooperativeStickyAssignor] - + +=== listener container === +=== listener container === +ackMode BATCH +groupId orders-basic +concurrency +[INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 3.828 s -- in com.ankurm.kafkabasics.EffectiveConfigTest +[INFO] Running com.ankurm.kafkabasics.SerialisationTest +JacksonJsonSerializer -> {"orderId":"o-1","customerId":"c-1","amount":10.00,"placedAt":"2026-08-29T10:15:30Z"} +[INFO] Tests run: 4, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 0.227 s -- in com.ankurm.kafkabasics.SerialisationTest +[INFO] Running com.ankurm.kafkabasics.KeysAndPartitionsTest +=== key -> partition, 3 partitions === +key partition murmur2 & 0x7fffffff % 3 Math.abs(murmur2) % 3 +o-1 0 0 0 +o-2 0 0 2 +o-3 1 1 1 +o-4 1 1 1 +o-5 2 2 0 +o-6 1 1 1 +2026-08-29T09:44:16.725+05:30 INFO 67 --- [kafka-basics] [ntainer#0-0-C-1] com.ankurm.kafkabasics.OrderConsumer : received orderId=o-1 partition=0 offset=0 +2026-08-29T09:44:16.726+05:30 INFO 67 --- [kafka-basics] [ntainer#0-0-C-1] com.ankurm.kafkabasics.OrderConsumer : received orderId=o-2 partition=0 offset=1 +2026-08-29T09:44:16.726+05:30 INFO 67 --- [kafka-basics] [ntainer#0-0-C-1] com.ankurm.kafkabasics.OrderConsumer : received orderId=o-1 partition=0 offset=2 +2026-08-29T09:44:16.726+05:30 INFO 67 --- [kafka-basics] [ntainer#0-0-C-1] com.ankurm.kafkabasics.OrderConsumer : received orderId=o-3 partition=1 offset=0 +2026-08-29T09:44:16.726+05:30 INFO 67 --- [kafka-basics] [ntainer#0-0-C-1] com.ankurm.kafkabasics.OrderConsumer : received orderId=o-4 partition=1 offset=1 +2026-08-29T09:44:16.727+05:30 INFO 67 --- [kafka-basics] [ntainer#0-0-C-1] com.ankurm.kafkabasics.OrderConsumer : received orderId=o-6 partition=1 offset=2 +2026-08-29T09:44:16.727+05:30 INFO 67 --- [kafka-basics] [ntainer#0-0-C-1] com.ankurm.kafkabasics.OrderConsumer : received orderId=o-5 partition=2 offset=0 +2026-08-29T09:44:16.746+05:30 INFO 67 --- [kafka-basics] [ntainer#0-0-C-1] com.ankurm.kafkabasics.OrderConsumer : received orderId=e-1 partition=0 offset=3 +2026-08-29T09:44:16.749+05:30 INFO 67 --- [kafka-basics] [ntainer#0-0-C-1] com.ankurm.kafkabasics.OrderConsumer : received orderId=e-2 partition=1 offset=3 +[INFO] Tests run: 2, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 0.532 s -- in com.ankurm.kafkabasics.KeysAndPartitionsTest +[INFO] Running com.ankurm.kafkabasics.ManualAckTest +Bootstrap metadata: BootstrapMetadata(records=[ApiMessageAndVersion(FeatureLevelRecord(name='metadata.version', featureLevel=30) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='eligible.leader.replicas.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='group.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='share.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='streams.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='transaction.version', featureLevel=2) at version 0)], metadataVersionLevel=30, source=format command) +Formatting metadata directory /tmp/kafka-14668657738428164299/combined_0_0 with metadata.version 4.3-IV0. +2026-08-29T09:44:17.569+05:30 INFO 67 --- [kafka-basics] [ main] com.ankurm.kafkabasics.ManualAckTest : Starting ManualAckTest using Java 25.0.4.1 with PID 67 (started by tender-adoring-cori in /tmp/work/smd/kafka-basics) +2026-08-29T09:44:17.570+05:30 INFO 67 --- [kafka-basics] [ main] com.ankurm.kafkabasics.ManualAckTest : The following 1 profile is active: "test" +2026-08-29T09:44:17.780+05:30 INFO 67 --- [kafka-basics] [ main] com.ankurm.kafkabasics.ManualAckTest : Started ManualAckTest in 0.503 seconds (process running for 6.368) +2026-08-29T09:44:18.617+05:30 INFO 67 --- [kafka-basics] [ntainer#0-0-C-1] com.ankurm.kafkabasics.OrderConsumer : received orderId=o-1 partition=0 offset=0 +[INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 1.919 s -- in com.ankurm.kafkabasics.ManualAckTest +[INFO] +[INFO] Results: +[INFO] +[INFO] Tests run: 8, Failures: 0, Errors: 0, Skipped: 0 +[INFO] +[INFO] ------------------------------------------------------------------------ +[INFO] BUILD SUCCESS +[INFO] ------------------------------------------------------------------------ +[INFO] Total time: 9.869 s +[INFO] Finished at: 2026-08-29T09:44:20+05:30 +[INFO] ------------------------------------------------------------------------ diff --git a/kafka-basics/docs/output/key-to-partition.txt b/kafka-basics/docs/output/key-to-partition.txt new file mode 100644 index 0000000..28f7cb3 --- /dev/null +++ b/kafka-basics/docs/output/key-to-partition.txt @@ -0,0 +1,7 @@ +key partition murmur2 & 0x7fffffff % 3 Math.abs(murmur2) % 3 +o-1 0 0 0 +o-2 0 0 2 +o-3 1 1 1 +o-4 1 1 1 +o-5 2 2 0 +o-6 1 1 1 diff --git a/kafka-basics/docs/output/serialised-payload.txt b/kafka-basics/docs/output/serialised-payload.txt new file mode 100644 index 0000000..fa24f53 --- /dev/null +++ b/kafka-basics/docs/output/serialised-payload.txt @@ -0,0 +1 @@ +JacksonJsonSerializer -> {"orderId":"o-1","customerId":"c-1","amount":10.00,"placedAt":"2026-08-29T10:15:30Z"} diff --git a/kafka-basics/docs/output/tests.txt b/kafka-basics/docs/output/tests.txt new file mode 100644 index 0000000..158d462 --- /dev/null +++ b/kafka-basics/docs/output/tests.txt @@ -0,0 +1,4 @@ +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 3.828 s -- in com.ankurm.kafkabasics.EffectiveConfigTest +Tests run: 4, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 0.227 s -- in com.ankurm.kafkabasics.SerialisationTest +Tests run: 2, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 0.532 s -- in com.ankurm.kafkabasics.KeysAndPartitionsTest +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 1.919 s -- in com.ankurm.kafkabasics.ManualAckTest diff --git a/kafka-basics/pom.xml b/kafka-basics/pom.xml new file mode 100644 index 0000000..0567a6e --- /dev/null +++ b/kafka-basics/pom.xml @@ -0,0 +1,84 @@ + + 4.0.0 + + + + org.springframework.boot + spring-boot-starter-parent + 4.1.1 + + + + com.ankurm + kafka-basics + 1.0 + jar + + + 25 + UTF-8 + + + + + org.springframework.boot + spring-boot-starter + + + + org.springframework.boot + spring-boot-starter-kafka + + + + org.springframework.boot + spring-boot-starter-jackson + + + + org.springframework.boot + spring-boot-starter-test + test + + + + org.springframework.kafka + spring-kafka-test + test + + + org.springframework.boot + spring-boot-testcontainers + test + + + + org.testcontainers + testcontainers-kafka + test + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + diff --git a/kafka-basics/scripts/run-all.sh b/kafka-basics/scripts/run-all.sh new file mode 100755 index 0000000..48a4726 --- /dev/null +++ b/kafka-basics/scripts/run-all.sh @@ -0,0 +1,15 @@ +#!/usr/bin/env bash +# Regenerate every file under docs/output/. Needs no Docker and no local Kafka: the tests start +# a real broker in KRaft mode inside the JVM. +set -eu +cd "$(dirname "$0")/.." +mvn -B test 2>&1 | tee /tmp/kafka-basics-test.log > /dev/null + +sed -n '/=== producer ===/,/=== listener container ===/p' /tmp/kafka-basics-test.log \ + > docs/output/effective-config.txt +sed -n '/=== listener container ===/,/^$/p' /tmp/kafka-basics-test.log >> docs/output/effective-config.txt + +sed -n '/^key *partition/,/^o-6/p' /tmp/kafka-basics-test.log > docs/output/key-to-partition.txt +grep 'JacksonJsonSerializer ->' /tmp/kafka-basics-test.log > docs/output/serialised-payload.txt +grep -E 'Tests run:.*in com\.ankurm' /tmp/kafka-basics-test.log | sed 's/^\[INFO\] //' > docs/output/tests.txt +echo "regenerated:"; ls -1 docs/output/ diff --git a/kafka-basics/src/main/java/com/ankurm/kafkabasics/KafkaBasicsApplication.java b/kafka-basics/src/main/java/com/ankurm/kafkabasics/KafkaBasicsApplication.java new file mode 100644 index 0000000..ae2c3c5 --- /dev/null +++ b/kafka-basics/src/main/java/com/ankurm/kafkabasics/KafkaBasicsApplication.java @@ -0,0 +1,20 @@ +package com.ankurm.kafkabasics; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +/** + * An order-events pipeline, small enough to read in one sitting. + * + *

Everything the companion article claims was produced by the tests in {@code src/test}, + * against a real Kafka broker running in KRaft mode. See the module README for how to point it + * at a Testcontainers broker instead. + */ +@SpringBootApplication +public class KafkaBasicsApplication { + + public static void main(String[] args) { + SpringApplication.run(KafkaBasicsApplication.class, args); + } + +} diff --git a/kafka-basics/src/main/java/com/ankurm/kafkabasics/OrderConsumer.java b/kafka-basics/src/main/java/com/ankurm/kafkabasics/OrderConsumer.java new file mode 100644 index 0000000..3362da6 --- /dev/null +++ b/kafka-basics/src/main/java/com/ankurm/kafkabasics/OrderConsumer.java @@ -0,0 +1,48 @@ +package com.ankurm.kafkabasics; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +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.Component; + +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +/** + * The consuming side. Note what is NOT here: no polling loop, no offset commit, no rebalance + * listener. The container owns all of that, which is the actual value Spring Kafka adds. + * + * @see docs/05-consuming.md + */ +@Component +public class OrderConsumer { + + private static final Logger log = LoggerFactory.getLogger(OrderConsumer.class); + + private final List received = new CopyOnWriteArrayList<>(); + + /** What arrived, with the metadata the tests assert on. */ + public record Received(OrderEvent event, int partition, long offset, String key) { + } + + @KafkaListener(topics = OrderProducer.TOPIC, groupId = "orders-basic") + public void onOrder(@Payload OrderEvent event, + @Header(KafkaHeaders.RECEIVED_PARTITION) int partition, + @Header(KafkaHeaders.OFFSET) long offset, + @Header(name = KafkaHeaders.RECEIVED_KEY, required = false) String key) { + log.info("received orderId={} partition={} offset={}", event.orderId(), partition, offset); + this.received.add(new Received(event, partition, offset, key)); + } + + public List received() { + return this.received; + } + + public void clear() { + this.received.clear(); + } + +} diff --git a/kafka-basics/src/main/java/com/ankurm/kafkabasics/OrderEvent.java b/kafka-basics/src/main/java/com/ankurm/kafkabasics/OrderEvent.java new file mode 100644 index 0000000..d456454 --- /dev/null +++ b/kafka-basics/src/main/java/com/ankurm/kafkabasics/OrderEvent.java @@ -0,0 +1,22 @@ +package com.ankurm.kafkabasics; + +import java.math.BigDecimal; +import java.time.Instant; + +/** + * The payload. A record works as a Kafka value with no annotations at all, because Jackson 3 + * handles records natively — but note that it needs a canonical constructor the + * deserializer can call, which is the one thing a record gives you for free and a Lombok + * {@code @Builder}-only class does not. + * + * @param orderId the partition key. Same customer, same order, same partition, same order of + * delivery — see docs/04-keys-and-partitions.md + */ +public record OrderEvent(String orderId, String customerId, BigDecimal amount, Instant placedAt) { + + public static OrderEvent of(String orderId, String customerId, String amount) { + return new OrderEvent(orderId, customerId, new BigDecimal(amount), + Instant.parse("2026-08-29T10:15:30Z")); + } + +} diff --git a/kafka-basics/src/main/java/com/ankurm/kafkabasics/OrderProducer.java b/kafka-basics/src/main/java/com/ankurm/kafkabasics/OrderProducer.java new file mode 100644 index 0000000..b0bf7b1 --- /dev/null +++ b/kafka-basics/src/main/java/com/ankurm/kafkabasics/OrderProducer.java @@ -0,0 +1,66 @@ +package com.ankurm.kafkabasics; + +import org.apache.kafka.clients.producer.ProducerRecord; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.support.SendResult; +import org.springframework.stereotype.Component; + +import java.util.concurrent.CompletableFuture; + +/** + * Three ways to send, and only two of them tell you when the send failed. + * + * @see docs/02-producing.md + */ +@Component +public class OrderProducer { + + private static final Logger log = LoggerFactory.getLogger(OrderProducer.class); + + static final String TOPIC = "orders"; + + private final KafkaTemplate template; + + public OrderProducer(KafkaTemplate template) { + this.template = template; + } + + /** + * Fire and forget. {@code send} returns a {@code CompletableFuture} and this method throws + * it away, so a broker-side rejection is invisible here: the method returns normally, the + * message never lands, and nothing in your logs says so unless you have the producer's own + * logger turned up. This is the single most common way to lose messages in Spring Kafka. + */ + public void sendAndForget(OrderEvent event) { + this.template.send(TOPIC, event.orderId(), event); + } + + /** Asynchronous, but the outcome is handled. This is the shape you want by default. */ + public CompletableFuture> send(OrderEvent event) { + CompletableFuture> future = + this.template.send(TOPIC, event.orderId(), event); + future.whenComplete((result, ex) -> { + if (ex != null) { + log.error("send failed for orderId={}", event.orderId(), ex); + } + else { + log.info("sent orderId={} to {}-{}@{}", event.orderId(), + result.getRecordMetadata().topic(), result.getRecordMetadata().partition(), + result.getRecordMetadata().offset()); + } + }); + return future; + } + + /** + * Synchronous. Correct when the caller must not proceed unless the write is durable, and + * expensive for exactly that reason: it blocks a thread for a network round trip plus the + * replication acknowledgement. + */ + public SendResult sendAndWait(OrderEvent event) throws Exception { + return this.template.send(new ProducerRecord<>(TOPIC, event.orderId(), event)).get(); + } + +} diff --git a/kafka-basics/src/main/resources/application.yaml b/kafka-basics/src/main/resources/application.yaml new file mode 100644 index 0000000..111b44d --- /dev/null +++ b/kafka-basics/src/main/resources/application.yaml @@ -0,0 +1,33 @@ +spring: + application: + name: kafka-basics + main: + banner-mode: off + kafka: + bootstrap-servers: localhost:9092 + producer: + key-serializer: org.apache.kafka.common.serialization.StringSerializer + # JacksonJsonSerializer, not JsonSerializer. Spring Kafka 4.1 ships both: JsonSerializer is + # the Jackson 2 one (com.fasterxml.jackson.databind.ObjectMapper) and JacksonJsonSerializer + # is the Jackson 3 one (tools.jackson.databind.json.JsonMapper). Boot 4 is a Jackson 3 + # application, and the Jackson 2 serializer's default mapper has no JSR-310 module, so a + # payload containing an Instant fails at send time. See docs/03-serialisation.md. + value-serializer: org.springframework.kafka.support.serializer.JacksonJsonSerializer + consumer: + group-id: orders-basic + # 'earliest' is NOT the Kafka default. The client default is 'latest', which means a + # brand-new consumer group sees nothing that was produced before it started - the single + # most common "my listener never fires" cause. See docs/05-consuming.md. + auto-offset-reset: earliest + key-deserializer: org.apache.kafka.common.serialization.StringDeserializer + value-deserializer: org.springframework.kafka.support.serializer.JacksonJsonDeserializer + properties: + # Without this the deserializer refuses the class named in the __TypeId__ header and + # the record is unreadable. There is no sensible default here on purpose: honouring an + # arbitrary class name from a message header is a deserialization gadget. + spring.json.trusted.packages: com.ankurm.kafkabasics +logging: + level: + root: WARN + com.ankurm: INFO + org.apache.kafka: ERROR diff --git a/kafka-basics/src/test/java/com/ankurm/kafkabasics/EffectiveConfigTest.java b/kafka-basics/src/test/java/com/ankurm/kafkabasics/EffectiveConfigTest.java new file mode 100644 index 0000000..31f41ae --- /dev/null +++ b/kafka-basics/src/test/java/com/ankurm/kafkabasics/EffectiveConfigTest.java @@ -0,0 +1,127 @@ +package com.ankurm.kafkabasics; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.config.ConfigDef; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.kafka.config.KafkaListenerEndpointRegistry; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.ProducerFactory; +import org.springframework.kafka.listener.MessageListenerContainer; +import org.springframework.kafka.test.context.EmbeddedKafka; +import org.springframework.test.context.ActiveProfiles; + +import java.lang.reflect.Field; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Prints three columns for each setting that matters: the kafka-clients default, whatever Spring + * Boot put on top of it, and therefore the effective value. Defaults move between releases and + * this is cheaper than remembering which. + * + *

The client defaults are read out of {@code ProducerConfig}/{@code ConsumerConfig}'s own + * {@code ConfigDef} by reflection, so they are the real ones for the kafka-clients version on + * this classpath rather than the ones the documentation happened to describe. + * + * @see docs/06-acknowledgement.md + */ +@SpringBootTest +@ActiveProfiles("test") +@EmbeddedKafka(topics = { "orders" }, partitions = 3) +class EffectiveConfigTest { + + @Autowired + ProducerFactory producerFactory; + + @Autowired + ConsumerFactory consumerFactory; + + @Autowired + KafkaListenerEndpointRegistry registry; + + private static final List PRODUCER_KEYS = List.of(ProducerConfig.ACKS_CONFIG, + ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, ProducerConfig.RETRIES_CONFIG, + ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, ProducerConfig.LINGER_MS_CONFIG, + ProducerConfig.BATCH_SIZE_CONFIG, ProducerConfig.COMPRESSION_TYPE_CONFIG, + ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG); + + private static final List CONSUMER_KEYS = List.of(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, + ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, ConsumerConfig.MAX_POLL_RECORDS_CONFIG, + ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, + ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, ConsumerConfig.ISOLATION_LEVEL_CONFIG, + ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG); + + @Test + void printEffectiveConfiguration() throws Exception { + table("producer", clientDefaults(ProducerConfig.class), PRODUCER_KEYS, + this.producerFactory.getConfigurationProperties()); + System.out.println(); + table("consumer", clientDefaults(ConsumerConfig.class), CONSUMER_KEYS, + this.consumerFactory.getConfigurationProperties()); + + Map producer = this.producerFactory.getConfigurationProperties(); + Map consumer = this.consumerFactory.getConfigurationProperties(); + + System.out.println(); + System.out.println("=== listener container ==="); + for (MessageListenerContainer container : this.registry.getListenerContainers()) { + System.out.printf("%-46s %s%n", "ackMode", container.getContainerProperties().getAckMode()); + System.out.printf("%-46s %s%n", "groupId", container.getGroupId()); + System.out.printf("%-46s %s%n", "concurrency", container.getContainerProperties().getClientId()); + } + + // Boot sets NOTHING on the producer beyond serializers and bootstrap servers. Durability + // therefore comes entirely from the kafka-clients defaults, which since Kafka 3.0 are + // acks=all and enable.idempotence=true. Absent is not unsafe here - but it does mean a + // property set in a Kafka 2.x-era runbook will change behaviour when you remove it. + assertThat(producer).doesNotContainKeys(ProducerConfig.ACKS_CONFIG, + ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, ProducerConfig.RETRIES_CONFIG); + + // The kafka-clients default for enable.auto.commit is TRUE, and Boot does not override it. + // The container does - but NOT by touching this shared factory. ListenerConsumer's + // determineAutoCommit checks whether the ConsumerFactory config contains the key and, when + // it does not, calls setProperty("enable.auto.commit", "false") on the per-container + // Properties handed to createConsumer. So the factory map never shows it, before or after + // the containers start, and reading the factory to find out whether auto-commit is on + // gives you the wrong answer. + assertThat(defaultOf(ConsumerConfig.class, ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG)) + .isEqualTo(true); + assertThat(this.consumerFactory.getConfigurationProperties()) + .doesNotContainKey(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG); + // And the public accessor on the factory agrees with the factory, not with reality: + // isAutoCommit() reads the same absent key and falls back to the client default, so it + // answers TRUE for a stock Boot 4.1 application in which no consumer auto-commits. + assertThat(this.consumerFactory.isAutoCommit()).isTrue(); + + // It DOES set isolation.level, which is the one place Boot has an opinion. + assertThat(consumer).containsEntry(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_uncommitted"); + } + + private void table(String title, Map defaults, List keys, + Map spring) { + System.out.println("=== " + title + " ==="); + System.out.printf("%-46s %-28s %-28s%n", "property", "kafka-clients default", "set by Spring Boot"); + System.out.println("-".repeat(104)); + for (String key : keys) { + System.out.printf("%-46s %-28s %-28s%n", key, String.valueOf(defaults.get(key)), + spring.containsKey(key) ? spring.get(key) : "-"); + } + } + + private static Map clientDefaults(Class configClass) throws Exception { + Field field = configClass.getDeclaredField("CONFIG"); + field.setAccessible(true); + ConfigDef configDef = (ConfigDef) field.get(null); + return configDef.defaultValues(); + } + + private static Object defaultOf(Class configClass, String key) throws Exception { + return clientDefaults(configClass).get(key); + } + +} diff --git a/kafka-basics/src/test/java/com/ankurm/kafkabasics/KeysAndPartitionsTest.java b/kafka-basics/src/test/java/com/ankurm/kafkabasics/KeysAndPartitionsTest.java new file mode 100644 index 0000000..a6fb048 --- /dev/null +++ b/kafka-basics/src/test/java/com/ankurm/kafkabasics/KeysAndPartitionsTest.java @@ -0,0 +1,95 @@ +package com.ankurm.kafkabasics; + +import org.apache.kafka.common.utils.Utils; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.kafka.support.SendResult; +import org.springframework.kafka.test.context.EmbeddedKafka; +import org.springframework.test.context.ActiveProfiles; + +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +/** + * Where a record lands, and why the answer is worth knowing before you pick a key. + * + *

Ordering in Kafka is per partition, never per topic. So the key is not a label — it is + * the ordering guarantee, and choosing it is the most consequential design decision in a + * producer. + * + * @see docs/04-keys-and-partitions.md + */ +@SpringBootTest +@ActiveProfiles("test") +@EmbeddedKafka(topics = { "orders" }, partitions = 3) +class KeysAndPartitionsTest { + + @Autowired + OrderProducer producer; + + @Autowired + OrderConsumer consumer; + + @Test + void theSameKeyAlwaysLandsOnTheSamePartition() throws Exception { + Map placement = new LinkedHashMap<>(); + for (String orderId : List.of("o-1", "o-2", "o-3", "o-4", "o-5", "o-6")) { + SendResult result = + this.producer.sendAndWait(OrderEvent.of(orderId, "c-1", "10.00")); + placement.put(orderId, result.getRecordMetadata().partition()); + } + + System.out.println("=== key -> partition, 3 partitions ==="); + System.out.printf("%-6s %-11s %-24s %s%n", "key", "partition", "murmur2 & 0x7fffffff % 3", + "Math.abs(murmur2) % 3"); + placement.forEach((key, partition) -> System.out.printf("%-6s %-11d %-24d %d%n", key, partition, + partitionFor(key), Math.abs(Utils.murmur2(key.getBytes(StandardCharsets.UTF_8))) % 3)); + + // The default partitioner is murmur2 of the serialized key, modulo the partition count. + // It is deterministic and it is not a hash you can change your mind about later: + // ADDING PARTITIONS REPARTITIONS EVERY KEY, which breaks per-key ordering across the + // boundary for anything still in flight. + placement.forEach((key, partition) -> assertThat(partition).isEqualTo(partitionFor(key))); + + // Sending the same key again lands in the same place. + SendResult again = this.producer.sendAndWait(OrderEvent.of("o-1", "c-9", "99.00")); + assertThat(again.getRecordMetadata().partition()).isEqualTo(placement.get("o-1")); + } + + /** + * The real formula. Note {@code & 0x7fffffff} and not {@code Math.abs}: they disagree + * whenever murmur2 returns a negative value, because masking the sign bit is not the same + * number as negating. Writing it with {@code Math.abs} reproduces the broker's placement for + * roughly half of all keys, which is the worst possible failure mode for a test. + */ + private static int partitionFor(String key) { + return (Utils.murmur2(key.getBytes(StandardCharsets.UTF_8)) & 0x7fffffff) % 3; + } + + @Test + void everythingProducedIsConsumedWithItsMetadata() { + this.consumer.clear(); + this.producer.sendAndForget(OrderEvent.of("e-1", "c-1", "1.00")); + this.producer.sendAndForget(OrderEvent.of("e-2", "c-2", "2.00")); + + await().atMost(Duration.ofSeconds(20)) + .until(() -> this.consumer.received().stream() + .map((r) -> r.event().orderId()).toList().containsAll(List.of("e-1", "e-2"))); + + assertThat(this.consumer.received()) + .extracting((r) -> r.event().orderId()).contains("e-1", "e-2"); + // The key arrives as a header, deserialized by the KEY deserializer, and it is the + // producer's key - not anything derived from the payload. + assertThat(this.consumer.received()).allSatisfy( + (r) -> assertThat(r.key()).isEqualTo(r.event().orderId())); + assertThat(this.consumer.received()).allSatisfy((r) -> assertThat(r.offset()).isGreaterThanOrEqualTo(0)); + } + +} diff --git a/kafka-basics/src/test/java/com/ankurm/kafkabasics/ManualAckTest.java b/kafka-basics/src/test/java/com/ankurm/kafkabasics/ManualAckTest.java new file mode 100644 index 0000000..c49155c --- /dev/null +++ b/kafka-basics/src/test/java/com/ankurm/kafkabasics/ManualAckTest.java @@ -0,0 +1,85 @@ +package com.ankurm.kafkabasics; + +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.kafka.config.KafkaListenerEndpointRegistry; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.listener.ContainerProperties.AckMode; +import org.springframework.kafka.support.Acknowledgment; +import org.springframework.kafka.test.context.EmbeddedKafka; +import org.springframework.boot.test.context.TestConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.test.context.ActiveProfiles; +import org.springframework.test.context.TestPropertySource; + +import java.time.Duration; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +/** + * Manual acknowledgement, and the guard that stands behind it. + * + *

{@code ConsumerFactory.isAutoCommit()} answers {@code true} on a stock configuration + * (see {@link EffectiveConfigTest}), so the interesting question is whether asking for + * {@code AckMode.MANUAL} trips Spring Kafka's assertion. It does not, because + * {@code determineAutoCommit} sets the per-container property to {@code false} first and the + * check that follows uses that, not the factory's opinion. + * + * @see docs/06-acknowledgement.md + */ +@SpringBootTest +@ActiveProfiles("test") +@TestPropertySource(properties = { + "spring.kafka.listener.ack-mode=MANUAL", + "spring.kafka.consumer.group-id=orders-manual" }) +@EmbeddedKafka(topics = { "orders" }, partitions = 1) +class ManualAckTest { + + static final List acked = new CopyOnWriteArrayList<>(); + + @TestConfiguration + static class Listeners { + + @Bean + ManualListener manualListener() { + return new ManualListener(); + } + + } + + static class ManualListener { + + @KafkaListener(topics = "orders", groupId = "orders-manual") + void onOrder(OrderEvent event, Acknowledgment acknowledgment) { + acked.add(event.orderId()); + // Nothing is committed until this line runs. If the process dies above it, the + // record is redelivered - which is the whole point, and also why an Acknowledgment + // you forget to call silently stalls the partition once max.poll.records is reached. + acknowledgment.acknowledge(); + } + + } + + @Autowired + KafkaTemplate template; + + @Autowired + KafkaListenerEndpointRegistry registry; + + @Test + void manualAckModeIsAcceptedAndUsed() { + assertThat(this.registry.getListenerContainers()) + .allSatisfy((container) -> assertThat(container.getContainerProperties().getAckMode()) + .isEqualTo(AckMode.MANUAL)); + + this.template.send("orders", "o-1", OrderEvent.of("o-1", "c-1", "10.00")); + await().atMost(Duration.ofSeconds(20)).until(() -> acked.contains("o-1")); + assertThat(acked).contains("o-1"); + } + +} diff --git a/kafka-basics/src/test/java/com/ankurm/kafkabasics/SerialisationTest.java b/kafka-basics/src/test/java/com/ankurm/kafkabasics/SerialisationTest.java new file mode 100644 index 0000000..d78aaef --- /dev/null +++ b/kafka-basics/src/test/java/com/ankurm/kafkabasics/SerialisationTest.java @@ -0,0 +1,91 @@ +package com.ankurm.kafkabasics; + +import org.apache.kafka.common.errors.SerializationException; +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.header.internals.RecordHeaders; +import org.junit.jupiter.api.Test; +import org.springframework.kafka.support.serializer.JacksonJsonDeserializer; +import org.springframework.kafka.support.serializer.JacksonJsonSerializer; +import org.springframework.kafka.support.serializer.JsonSerializer; + +import java.nio.charset.StandardCharsets; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; + +/** + * The two JSON serializer families Spring Kafka 4.1 ships, side by side. No broker needed: a + * Serializer is a function from object to bytes and can be called directly, which is the + * cheapest possible way to settle a serialisation question. + * + * @see docs/03-serialisation.md + */ +class SerialisationTest { + + private static final OrderEvent EVENT = OrderEvent.of("o-1", "c-1", "10.00"); + + @Test + void theJackson2SerializerCannotWriteAnInstant() { + // JsonSerializer is the Jackson 2 one. Its default ObjectMapper has no JSR-310 module, + // and Jackson 2.21 refuses java.time types rather than guessing at a representation. + // This is what you get by following any pre-Boot-4 tutorial. + try (JsonSerializer serializer = new JsonSerializer<>()) { + assertThatExceptionOfType(SerializationException.class) + .isThrownBy(() -> serializer.serialize("orders", EVENT)) + .withMessageContaining("Can't serialize data") + .withStackTraceContaining("Java 8 date/time type `java.time.Instant` not supported by default"); + } + } + + @Test + void theJackson3SerializerWritesItWithNoConfiguration() { + try (JacksonJsonSerializer serializer = new JacksonJsonSerializer<>()) { + String json = new String(serializer.serialize("orders", EVENT), StandardCharsets.UTF_8); + System.out.println("JacksonJsonSerializer -> " + json); + assertThat(json).contains("\"orderId\":\"o-1\"").contains("\"placedAt\":"); + // BigDecimal survives as a number, not a string, and keeps its scale. + assertThat(json).contains("\"amount\":10.00"); + } + } + + @Test + void theDeserializerRefusesAnUntrustedClassNamedInTheTypeHeader() { + Headers headers = new RecordHeaders(); + byte[] bytes; + try (JacksonJsonSerializer serializer = new JacksonJsonSerializer<>()) { + // The three-argument overload is the one that writes __TypeId__. That header is how the + // consumer learns which class to build, and it is also why trusted packages exist: + // instantiating a class named by an inbound message is a deserialization gadget. + bytes = serializer.serialize("orders", headers, EVENT); + } + assertThat(headers.lastHeader("__TypeId__")).isNotNull(); + assertThat(new String(headers.lastHeader("__TypeId__").value(), StandardCharsets.UTF_8)) + .isEqualTo(OrderEvent.class.getName()); + + try (JacksonJsonDeserializer deserializer = new JacksonJsonDeserializer<>()) { + deserializer.configure(Map.of(), false); + assertThatExceptionOfType(Exception.class) + .isThrownBy(() -> deserializer.deserialize("orders", headers, bytes)) + .withMessageContaining("not in the trusted packages"); + } + } + + @Test + void aRoundTripThroughTheJackson3PairIsLossless() { + byte[] bytes; + try (JacksonJsonSerializer serializer = new JacksonJsonSerializer<>()) { + bytes = serializer.serialize("orders", EVENT); + } + try (JacksonJsonDeserializer deserializer = new JacksonJsonDeserializer<>()) { + deserializer.configure(Map.of(JacksonJsonDeserializer.VALUE_DEFAULT_TYPE, + OrderEvent.class.getName(), JacksonJsonDeserializer.TRUSTED_PACKAGES, + "com.ankurm.kafkabasics"), false); + OrderEvent back = deserializer.deserialize("orders", bytes); + assertThat(back).isEqualTo(EVENT); + // BigDecimal scale survives the round trip. It would not if amount were a double. + assertThat(back.amount().scale()).isEqualTo(2); + } + } + +} diff --git a/kafka-basics/src/test/java/com/ankurm/kafkabasics/TestcontainersConfiguration.java b/kafka-basics/src/test/java/com/ankurm/kafkabasics/TestcontainersConfiguration.java new file mode 100644 index 0000000..0a1ce75 --- /dev/null +++ b/kafka-basics/src/test/java/com/ankurm/kafkabasics/TestcontainersConfiguration.java @@ -0,0 +1,45 @@ +package com.ankurm.kafkabasics; + +import org.springframework.boot.test.context.TestConfiguration; +import org.springframework.boot.testcontainers.service.connection.ServiceConnection; +import org.springframework.context.annotation.Bean; +import org.testcontainers.kafka.KafkaContainer; +import org.testcontainers.utility.DockerImageName; + +/** + * The Testcontainers route. Import this instead of {@code @EmbeddedKafka} when you want the same + * broker image your production cluster runs, or when you are testing something the in-process + * broker does not model (real network partitions, TLS, SASL, quotas). + * + *

{@code @ServiceConnection} is what makes this ergonomic: it registers the container's + * bootstrap servers as the application's, so there is no + * {@code @DynamicPropertySource} block and no {@code spring.kafka.bootstrap-servers} to keep in + * sync. It replaced that boilerplate in Boot 3.1 and is the only shape worth writing now. + * + *

Two coordinates matter and both changed recently: + *

    + *
  • the Maven artifact is {@code org.testcontainers:testcontainers-kafka}, not + * {@code org.testcontainers:kafka}, which stopped at 1.21.4
    • + *
  • the class is {@code org.testcontainers.kafka.KafkaContainer} (Apache Kafka, + * KRaft, no ZooKeeper), not {@code org.testcontainers.containers.KafkaContainer}
    • + *
+ * + *

The transcripts under {@code docs/output/} were NOT produced by this class — they came + * from the in-process KRaft broker, because the machine that regenerates them has no Docker + * daemon. Both paths run the same tests. + * + * @see docs/07-testing.md + */ +@TestConfiguration(proxyBeanMethods = false) +public class TestcontainersConfiguration { + + // @RestartScope from spring-boot-devtools is worth adding here if you use `bootTestRun`: + // it keeps the container alive across devtools restarts. It needs the devtools dependency, + // which this module deliberately does not have. + @Bean + @ServiceConnection + KafkaContainer kafkaContainer() { + return new KafkaContainer(DockerImageName.parse("apache/kafka:4.1.0")); + } + +} diff --git a/kafka-basics/src/test/resources/application-test.yaml b/kafka-basics/src/test/resources/application-test.yaml new file mode 100644 index 0000000..9981b8e --- /dev/null +++ b/kafka-basics/src/test/resources/application-test.yaml @@ -0,0 +1,4 @@ +spring: + kafka: + # Set by @EmbeddedKafka: the in-process KRaft broker picks a random port. + bootstrap-servers: ${spring.embedded.kafka.brokers}