Apache Kafka Streams is a client library for building applications and microservices, where the input and output data are stored in Kafka clusters. It allows for stateful and stateless transformations, aggregations, and joins on streaming data. Kafka Streams manages much of the complexity of dealing with distributed, scalable, and fault-tolerant applications internally. One crucial aspect of managing state in Kafka Streams is through the use of internal topics, primarily for state stores and repartitioning of data.

Cleanup Policy: delete vs. compact

Kafka topics can be configured with different cleanup policies to manage how old data is discarded. There are two main policies:

• delete: This policy will delete records once they reach a certain age or size. This is a common policy for topics that do not require a history of all data.
• compact: This policy is used in scenarios where the complete history of records for a specific key is not needed, but the latest value for each key is crucial. Compact retains at least the last known value for each key.

Issue with Setting Cleanup Policy to delete for Kafka Streams Internal Topics

Setting the cleanup policy to delete for Kafka Streams internal topics generally does not produce the intended behavior, especially for stateful applications. Kafka Streams uses these topics to store the state (KTables) or to shuffle data when re-partitioning is needed. Here's why using delete can be problematic:

1. Loss of State on Failure or Rebalance: Internal topics often store state that is critical for the correct functioning of the application. With the delete policy, if a failure or rebalance occurs, you might lose essential data that wasn't backed up elsewhere.
2. Incorrect Application Results: For stateful operations that rely on windowing or joining over a period, using delete can lead to missing records that are needed to compute results accurately.

Example of a Potential Fail Scenario with delete

Consider a Kafka Streams application that maintains a running total of sales per store in a KTable. The internal topic configured with the delete policy might remove older record data needed for the proper functioning of the app under certain operations like app restarts or rebalances.

StreamsBuilder builder = new StreamsBuilder();
KStream<String, Long> sales = builder.stream("sales-topic");
KTable<String, Long> salesByStore = sales.groupByKey().reduce(Long::sum);

In this simplistic example, should any message in the salesByStore internal topic be deleted prematurely because of the delete policy, the state recovery or application restart might show incorrect totals, as not all original sales records are available to reconstruct the KTable.

Best Practices and Solutions

To mitigate the issues associated with the wrong cleanup policy in Kafka Streams:

1. Default to compact: For most Kafka Streams use cases involving state, default to using the compact cleanup policy. This ensures that the latest state is always retained.
2. Explicitly Set Policies on Internal Topics: When creating streams, explicitly set the cleanup policy on internal topics, if the default does not suit your needs.

Example Code

1Properties streamsConfig = new Properties();
2streamsConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, "app-id");
3streamsConfig.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker:9092");
4streamsConfig.put(StreamsConfig.topic.cleanup.policy.config, TopicConfig.CLEANUP_POLICY_COMPACT);
5
6StreamsBuilder builder = new StreamsBuilder();
7builder.table("source-topic", Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("state-store-name")
8        .withKeySerde(Serdes.String())
9        .withValueSerde(Serdes.Long())
10        .withLoggingEnabled(Collections.singletonMap(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT)));

Summary Table

Conclusion

Setting internal topics cleanup policy to delete in Kafka Streams applications, particularly those with stateful operations, should be approached with caution. Misconfiguration can lead to incorrect data processing and potential loss of crucial state information. Adopting the compact policy by default and understanding the implications of different cleanup policies are key steps in optimizing Kafka Streams applications for reliability and correctness.