Apache Kafka is a distributed event streaming platform used extensively for building real-time data pipelines and streaming applications. One of its core components are "topics", where messages are categorized. Each Kafka topic can be configured with various policies, including those governing data retention. Retention policies are crucial for managing how long messages are stored on a topic before being deleted or compacted.

In a Spring application, configuring Kafka topic retention policies can be handled conveniently through the use of Spring Kafka, a project which extends the functionality of Kafka with Spring-friendly features. Below, we discuss how to configure a Kafka topic’s retention policy during its creation, specifically in the context of a Spring application.

Understanding Kafka Topic Retention Policy

Kafka stores messages in topics and the messages in these topics are retained according to the configured retention policy. The two main types of retention policies are:

1. Time-Based Retention (retention.ms): This policy determines how long messages are retained in a topic in terms of time. After this period, messages are eligible for deletion regardless of whether they have been consumed.
2. Size-Based Retention (retention.bytes): This policy limits the total size of messages retained on a topic. Once the size limit is reached, the oldest messages are deleted.
3. Compaction (cleanup.policy): This is different from deletion as it retains only the last message for each key in the topic.

Configuring Kafka Topics with Spring

Spring provides comprehensive support for Kafka through its Spring Kafka project. Configuring a Kafka topic with a specified retention policy involves defining these settings within the application's configuration class or within an external configuration file (like application.yml or application.properties).

Using @Configuration and @Bean

You can define Kafka topic configurations programmatically in a Spring configuration file using @Bean annotations:

1@Configuration
2public class KafkaTopicConfig {
3    
4    @Autowired
5    private KafkaAdmin kafkaAdmin;
6
7    @Bean
8    public NewTopic sampleTopic() {
9        return TopicBuilder.name("sample-topic")
10                .partitions(10)
11                .replicas(1)
12                .config(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(24 * 60 * 60 * 1000)) // 24 hours retention
13                .config(TopicConfig.CLEANUP_POLICY_CONFIG, "delete")
14                .build();
15    }
16}

In the above example, a topic named sample-topic is configured using the TopicBuilder API. The retention policy is set to delete messages after 24 hours.

Using application.yml or application.properties

Alternatively, you can configure these properties directly in the application's configuration files:

For application.yml:

1spring:
2  kafka:
3    producer:
4      bootstrap-servers: localhost:9092
5    properties:
6      retention.ms: 86400000  # 24 hours in milliseconds
7      cleanup.policy: delete

And in application.properties:

spring.kafka.producer.bootstrap-servers=localhost:9092
spring.kafka.properties.retention.ms=86400000  # 24 hours in milliseconds
spring.kafka.properties.cleanup.policy=delete

Key Takeaways

Here's a summary table of the key properties for Kafka topic retention configuration:

Conclusion

Configuring Kafka topic retention policies properly is essential for managing storage and ensuring that your application complies with data policies. With Spring Kafka, this process is streamlined, allowing developers to configure topics directly through code or configuration files, making management and adjustments easier as application requirements evolve.