Testing Kafka Applications

1. Using EmbeddedKafka

EmbeddedKafka is a testing utility provided by the spring-kafka-test module that allows developers to run a Kafka broker instance embedded within their test JVM. This is particularly useful for integration testing without the need to set up and manage external Kafka clusters.

Feature	Description	Usage
EmbeddedKafkaRule	JUnit rule that starts and stops an embedded Kafka broker.	`@Rule` annotation in JUnit tests.
EmbeddedKafkaExtension	JUnit 5 extension for managing the lifecycle of an embedded Kafka broker.	`@ExtendWith(EmbeddedKafkaExtension.class)`
KafkaEmbedded	Class that encapsulates the embedded Kafka broker instance.	Used to configure and manage Kafka server properties.

Example: Using EmbeddedKafka with JUnit 4

@Rule
public EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, 1, "test-topic");

@Test
public void testProducer() {
    // Use embedded Kafka for testing
}

Example: Using EmbeddedKafka with JUnit 5

@ExtendWith(EmbeddedKafkaExtension.class)
public class KafkaIntegrationTest {

    @Autowired
    private EmbeddedKafka kafka;

    @Test
    public void testConsumer() {
        // Use embedded Kafka for testing
    }
}

Note: Ensure that spring-kafka-test is included in your test dependencies.

Warning: Embedded Kafka is not suitable for performance or load testing due to its in-memory nature.

2. Creating Test Producer

Property	Value	Description
bootstrap.servers	`localhost:9092`	Initial list of brokers to connect to
key.serializer	`org.apache.kafka.common.serialization.StringSerializer`	Serializer for message keys
value.serializer	`org.apache.kafka.common.serialization.StringSerializer`	Serializer for message values
acks	`1`	Number of acknowledgments required from the leader

Example: Basic Test Producer Setup

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("acks", "1");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

Example: Sending a Test Message

ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", "key1", "Hello Kafka");
producer.send(record);

3. Creating Test Consumer

Property	Description	Default Value	Notes
bootstrap.servers	Comma-separated list of broker addresses	localhost:9092	Required for connecting to Kafka cluster
group.id	Consumer group identifier	None	Must be set for consumer groups
key.deserializer	Deserializer for message keys	org.apache.kafka.common.serialization.StringDeserializer	Must match key type in topic
value.deserializer	Deserializer for message values	org.apache.kafka.common.serialization.StringDeserializer	Must match value type in topic
enable.auto.commit	Enables automatic offset commits	true	Set to false for manual control
auto.offset.reset	Behavior when no initial offset is found	latest	earliest, latest, none

Example: Basic Test Consumer Setup

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-consumer-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("enable.auto.commit", "false");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("test-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
    }
}

Note: For testing, use enable.auto.commit=false to have full control over offset commits.

Warning: Ensure key.deserializer and value.deserializer match the actual data format in your topics.

4. Testing Streams Topology

Example: Basic Kafka Streams Test

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TestOutputTopic;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.Stores;

public class StreamsTest {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> source = builder.stream("input-topic");
        KStream<String, String> processed = source.mapValues(value -> value.toUpperCase());
        processed.to("output-topic");

        KafkaStreams streams = new KafkaStreams(builder.build(), getConfiguration());
        streams.start();

        TestInputTopic<String, String> inputTopic = streams.testInput("input-topic");
        TestOutputTopic<String, String> outputTopic = streams.testOutput("output-topic");

        inputTopic.pipeInput("key1", "hello");
        inputTopic.pipeInput("key2", "world");

        assertEquals(outputTopic.readValues(), Arrays.asList("HELLO", "WORLD"));
    }
}

Component	Description	Usage
KafkaStreams	Core class for running Kafka Streams applications	Used to configure and start the streams application in tests
StreamsBuilder	Builds the Kafka Streams topology	Used to define the stream processing logic in tests
TestInputTopic	Simulates input data for testing	Used to send test data into the streams application
TestOutputTopic	Collects output data for assertions	Used to read and verify the results of stream processing

Note: Ensure that the test environment has Kafka and ZooKeeper running, or use embedded versions for local testing.

Warning: Avoid using KafkaStreams in production with test configurations. Use separate configurations for testing and production.

Key Concepts

StreamsBuilder: Used to construct the Kafka Streams topology.
TestInputTopic: Simulates input data for testing.
TestOutputTopic: Collects and verifies output data from streams.

5. Mocking External Dependencies

Example: Using Mockito to Mock Kafka Producer

import org.apache.kafka.clients.producer.KafkaProducer;
import org.mockito.Mockito;

// Create a mock KafkaProducer
KafkaProducer<String, String> mockProducer = Mockito.mock(KafkaProducer.class);

// Configure behavior for send
Mockito.doReturn(new org.apache.kafka.clients.producer.RecordMetadata("topic", 0, 0, 0L, 0L))
        .when(mockProducer).send(Mockito.any(), Mockito.any());

// Use in test
mockProducer.send(new ProducerRecord<>("test-topic", "key", "value"));

Feature	Description	Notes
Mockito	`org.mockito.Mockito`	Popular mocking framework for Java
KafkaProducer	`org.apache.kafka.clients.producer.KafkaProducer`	Mockable Kafka producer class
KafkaConsumer	`org.apache.kafka.clients.consumer.KafkaConsumer`	Mockable Kafka consumer class

Note: When mocking Kafka, ensure that your test environment does not depend on actual broker connectivity.

Warning: Mocking Kafka producers and consumers can mask real issues in serialization or network configuration.

Best Practices for Mocking Kafka

Use Mockito or similar frameworks to simulate Kafka behavior.
Verify that messages are sent correctly without hitting the actual broker.
Test error handling by simulating producer failures.

6. Testing Serialization

Property	Value	Description
serializer.class	`org.apache.kafka.common.serialization.StringSerializer`	Serializes string values to bytes
serializer.class	`org.apache.kafka.common.serialization.ByteArraySerializer`	Serializes byte arrays directly
serializer.class	`org.apache.kafka.common.serialization.IntegerSerializer`	Serializes integer values to bytes
serializer.class	`org.apache.kafka.common.serialization.LongSerializer`	Serializes long values to bytes
serializer.class	`org.apache.kafka.common.serialization.DoubleSerializer`	Serializes double values to bytes
serializer.class	`org.apache.kafka.common.serialization.BooleanSerializer`	Serializes boolean values to bytes
serializer.class	`org.apache.kafka.common.serialization.FloatSerializer`	Serializes float values to bytes
serializer.class	`org.apache.kafka.common.serialization.ShortSerializer`	Serializes short values to bytes
serializer.class	`org.apache.kafka.common.serialization.ByteSerializer`	Serializes byte values to bytes

Example: String Serialization

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<String, String>("test-topic", "key", "value"));
producer.close();

Example: JSON Serialization

public class JsonSerializer<T> implements Serializer<T> {
    private final ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public byte[] serialize(String topic, T data) {
        try {
            return objectMapper.writeValueAsBytes(data);
        } catch (Exception e) {
            throw new SerializationException("Error serializing JSON", e);
        }
    }
}

Note: When testing, ensure that your test producers and consumers use the same serializers as in production to avoid serialization mismatches.

7. Testing Error Scenarios

Error Type	Description	Testing Approach	Tools/Methods
Producer Failures	Simulate producer errors like network issues or timeouts	Use mock producers to inject failures during message sending	`MockProducer`, `TestUtils`
Consumer Failures	Simulate consumer errors such as message processing failures or rebalancing issues	Inject exceptions in consumer logic to test error recovery and retry mechanisms	`ConsumerInterceptor`, `MockConsumer`
Broker Failures	Simulate broker unavailability or partition leader changes	Use embedded Kafka or Docker containers to simulate broker downtime	`KafkaServer`, `docker-compose`
Network Errors	Simulate network partitions or latency issues between clients and brokers	Use network simulation tools like `tc` or `netem`	`NetworkPartitionSimulator`, `tc`

Example: Simulating Producer Failures

import org.apache.kafka.clients.producer.MockProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

MockProducer<String, String> mockProducer = new MockProducer<>(true, new StringSerializer(), new StringSerializer());

// Simulate a send failure
mockProducer.sendError(new RuntimeException("Network timeout"));

Example: Testing Consumer Retry Logic

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

// Simulate processing failure
try {
    consumer.subscribe(Collections.singletonList("test-topic"));
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        throw new RuntimeException("Processing failed");
    }
} catch (Exception e) {
    // Retry logic or error handling
    System.out.println("Retrying after failure: " + e.getMessage());
}

Note: When simulating errors, ensure that your tests cover both transient and permanent failures to validate robustness.

8. Integration Testing

Example: Integration Test Setup with Embedded Kafka

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class KafkaIntegrationTest {
    private static final String BOOTSTRAP_SERVERS = "localhost:9092";
    private static final String TOPIC_NAME = "test-topic";

    public void testProducerAndConsumer() {
        Properties producerProps = new Properties();
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);
        ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, "key1", "message1");
        producer.send(record);
        producer.flush();
        producer.close();
    }
}

Component	Description	Notes
Kafka Embedded	Embedded Kafka server for testing	Useful for unit and integration tests
TestContainers	Containerized Kafka for integration tests	Provides real Kafka instances in containers
Kafka Streams Test Utilities	Utilities for testing Kafka Streams applications	Part of Kafka's test infrastructure

Note: Integration tests often involve setting up a Kafka cluster, producing messages, and verifying consumption or processing.

Best Practices for Integration Testing

Use embedded Kafka or TestContainers to simulate real environments
Ensure test isolation by using unique topic names per test
Validate message order and content in consumer tests
Test error handling and recovery scenarios

9. Performance Testing

Aspect	Description	Tools
Throughput	Measures the number of messages processed per second	Kafka Load Test, JMH
Latency	Time taken for a message to be produced and consumed	Kafka Producer/Consumer, JMeter
Resource Utilization	Monitoring CPU, Memory, Disk I/O	Systemd, Prometheus, Grafana

Example: Measuring Throughput

kafka-producer-perf-test.sh \
  --topic test-topic \
  --num-records 1000000 \
  --record-size 1024 \
  --throughput 10000 \
  --producer.config producer.properties

Example: Latency Measurement

kafka-consumer-perf-test.sh \
  --topic test-topic \
  --messages 1000000 \
  --consumer.config consumer.properties

Note: Use kafka-producer-perf-test.sh and kafka-consumer-perf-test.sh for basic performance testing. For more complex scenarios, consider using JMeter or custom load testing frameworks.

10. Load Testing

Example: Kafka Load Test Configuration

name: kafka-load-test
topics:
  - name: test-topic
    partitions: 10
    replication-factor: 3
producer:
  throughput: 10000
  message-size: 1024
consumer:
  concurrency: 5
  batch-size: 100
duration: 3600s

Configuration	Description	Default Value	Notes
name	Test scenario name	`default-test`	Used for reporting and logging
topics	List of topics to create	`[]`	Each topic has name, partitions, replication-factor
producer.throughput	Messages per second to produce	`1000`	Adjust based on system capacity
producer.message-size	Size of each message in bytes	`1024`	Simulates realistic payload sizes
consumer.concurrency	Number of concurrent consumers	`1`	Higher values increase throughput
duration	Duration of the test in seconds	`300s`	Use ISO 8601 format (e.g., 1h, 30m)

Note: Use kafka-producer-perf-test.sh and kafka-consumer-perf-test.sh for built-in load testing tools.

Warning: Ensure sufficient disk space and memory for large-scale load tests to avoid system instability.

Load Testing Best Practices

Start with small topic partitions and gradually scale up
Monitor broker CPU, memory, and disk I/O during tests
Use kafka-configs.sh to adjust topic configurations before testing
Validate consumer lag and throughput metrics post-test