Apache Kafka is a popular open-source stream-processing software platform developed by the Apache Software Foundation, written in Scala and Java. Designed to provide a unified, high-throughput, low-latency platform for handling real-time data feeds, Kafka is widely used in enterprise environments. One of its common needs is managing stored data, specifically deleting records from a Kafka topic.

Understanding Kafka Topic and Record Management

In Kafka, records (messages) are grouped in topics. In essence, a topic is a category or feed name to which records are published. Topics in Kafka are multi-subscriber, and they can have zero or many consumers that subscribe to the data written to them. The data within a topic is divided into partitions, which are distributed across multiple servers in the Kafka cluster to ensure redundancy and fault tolerance.

Why Delete Records?

Here are several reasons why one might want to delete records from a Kafka topic:

• Data retention policies: Organizations often have policies on how long data can be retained for compliance purposes.
• Disk space management: Removing obsolete or unnecessary data can help in managing disk space.
• Data correction: Removing incorrect or outdated information to maintain data integrity.

Record Deletion in Kafka

Kafka supports record deletion through two primary mechanisms:

1. Retention Policy: Kafka topics are configured with a retention policy that automatically deletes records older than a specified retention period or when the log reaches a certain size.
2. Tombstone Messages: For key-based topics, sending a message with a null value (tombstone) for a specific key marks that record for deletion.

1. Retention Policy Configuration

Each Kafka topic can be configured with retention settings that define how long records should be kept. The two primary configurations are:

• retention.ms: Controls how long records are retained based on time.
• retention.bytes: Controls the maximum size of the log before older records are deleted.

These settings can be adjusted at the topic level using the Kafka administrative client.

2. Deleting records using Tombstone messages

For key-based topics, records can be explicitly marked for deletion using a tombstone message. This is a record with the same key as the record to be deleted but with a null value. Kafka uses these tombstone messages to know that all records with that key should be removed upon log compaction.

Implementing Record Deletion Using Java

To manage Kafka topics and deletion using Java, you can use the Kafka AdminClient API for modifying topic settings, and the Producer API for sending tombstone messages. Below is an example demonstrating both methods.

Kafka AdminClient API Example

1import org.apache.kafka.clients.admin.*;
2
3Properties props = new Properties();
4props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
5
6try (AdminClient admin = KafkaAdminClient.create(props)) {
7    // Define new topic config with a shorter retention time
8    Map<String, String> configs = new HashMap<>();
9    configs.put("retention.ms", "10000");  // 10 seconds retention
10
11    // Create a new config entry
12    ConfigEntry retentionEntry = new ConfigEntry("retention.ms", "10000");
13    Config config = new Config(Arrays.asList(retentionEntry));
14
15    // Updating the topic configuration
16    AlterConfigsResult alterConfigsResult = admin.alterConfigs(Map.of(new ConfigResource(ConfigResource.Type.TOPIC, "your-topic-name"), config));
17    alterConfigsResult.all().get();
18} catch (Exception e) {
19    e.printStackTrace();
20}

Producer API for Sending Tombstone Messages

1import org.apache.kafka.clients.producer.*;
2
3Properties props = new Properties();
4props.put("bootstrap.servers", "localhost:9092");
5props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
6props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
7
8try (Producer<String, String> producer = new KafkaProducer<>(props)) {
9    // Sending a tombstone message for the key "key1"
10    producer.send(new ProducerRecord<>("your-topic-name", "key1", null));
11}

Summary Table

Conclusion

Kafka provides flexible mechanisms for deleting records, either through automated retention policies or manually through tombstone messages. Using the Java API, developers can manage these functions programmatically, integrating Kafka's data management features within larger applications or systems.