asmhatre/spring-messaging-demo
1
0
Fork 0
Code Activity

Add the kafka-basics module

Browse Source
This commit is contained in:
Ankur Mhatre
2026-08-29 09:47:25 +05:30
commit 3a682e496e
28 changed files with 1403 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

127
kafka-basics/src/test/java/com/ankurm/kafkabasics/EffectiveConfigTest.java Normal file
View File

@@ -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.
*
* <p>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 <a href="../../../../../docs/06-acknowledgement.md">docs/06-acknowledgement.md</a>
*/
@SpringBootTest
@ActiveProfiles("test")
@EmbeddedKafka(topics = { "orders" }, partitions = 3)
class EffectiveConfigTest {
@Autowired
ProducerFactory<String, OrderEvent> producerFactory;
@Autowired
ConsumerFactory<String, OrderEvent> consumerFactory;
@Autowired
KafkaListenerEndpointRegistry registry;
private static final List<String> 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<String> 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<String, Object> producer = this.producerFactory.getConfigurationProperties();
Map<String, Object> 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<String, Object> defaults, List<String> keys,
Map<String, Object> 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<String, Object> 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);
}
}

95
kafka-basics/src/test/java/com/ankurm/kafkabasics/KeysAndPartitionsTest.java Normal file
View File

@@ -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.
*
* <p>Ordering in Kafka is per partition, never per topic. So the key is not a label &mdash; it is
* the ordering guarantee, and choosing it is the most consequential design decision in a
* producer.
*
* @see <a href="../../../../../docs/04-keys-and-partitions.md">docs/04-keys-and-partitions.md</a>
*/
@SpringBootTest
@ActiveProfiles("test")
@EmbeddedKafka(topics = { "orders" }, partitions = 3)
class KeysAndPartitionsTest {
@Autowired
OrderProducer producer;
@Autowired
OrderConsumer consumer;
@Test
void theSameKeyAlwaysLandsOnTheSamePartition() throws Exception {
Map<String, Integer> placement = new LinkedHashMap<>();
for (String orderId : List.of("o-1", "o-2", "o-3", "o-4", "o-5", "o-6")) {
SendResult<String, OrderEvent> 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<String, OrderEvent> 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));
}
}

85
kafka-basics/src/test/java/com/ankurm/kafkabasics/ManualAckTest.java Normal file
View File

@@ -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.
*
* <p>{@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 <a href="../../../../../docs/06-acknowledgement.md">docs/06-acknowledgement.md</a>
*/
@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<String> 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<String, OrderEvent> 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");
}
}

91
kafka-basics/src/test/java/com/ankurm/kafkabasics/SerialisationTest.java Normal file
View File

@@ -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 <a href="../../../../../docs/03-serialisation.md">docs/03-serialisation.md</a>
*/
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<OrderEvent> 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<OrderEvent> 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<OrderEvent> 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<OrderEvent> 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<OrderEvent> serializer = new JacksonJsonSerializer<>()) {
bytes = serializer.serialize("orders", EVENT);
}
try (JacksonJsonDeserializer<OrderEvent> 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);
}
}
}

45
kafka-basics/src/test/java/com/ankurm/kafkabasics/TestcontainersConfiguration.java Normal file
View File

@@ -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).
*
* <p>{@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.
*
* <p>Two coordinates matter and both changed recently:
* <ul>
* <li>the Maven artifact is <b>{@code org.testcontainers:testcontainers-kafka}</b>, not
* {@code org.testcontainers:kafka}, which stopped at 1.21.4</li>
* <li>the class is <b>{@code org.testcontainers.kafka.KafkaContainer}</b> (Apache Kafka,
* KRaft, no ZooKeeper), not {@code org.testcontainers.containers.KafkaContainer}</li>
* </ul>
*
* <p>The transcripts under {@code docs/output/} were NOT produced by this class &mdash; 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 <a href="../../../../../docs/07-testing.md">docs/07-testing.md</a>
*/
@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"));
}
}