Apache Kafka kafka_2.13版本下如何配置与实践以永久保留KStream多窗口聚合的changelog主题数据

Great question! Let's break down exactly what you need to do to preserve all your windowed aggregation results indefinitely—even hourly windows from a year ago—while using Kafka Streams with kafka_2.13. Below are the key configuration rules and coding practices to make this happen:

By default, Kafka Streams automatically expires windows after a set retention period (usually 24 hours for tumbling windows). To prevent this, you need to explicitly set an infinite retention period for each windowed store:

When defining your aggregation, use the Materialized API to configure the underlying window store with the maximum possible retention. For example, for an hourly tumbling window:

// Hourly window aggregation
stream.groupByKey()
    .windowedBy(TumblingWindows.of(Duration.ofHours(1)))
    .aggregate(
        () -> 0L, // Initial value
        (key, value, aggregate) -> aggregate + value, // Aggregation logic
        Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("hourly-agg-store")
            .withKeySerde(Serdes.String())
            .withValueSerde(Serdes.Long())
            .withRetention(Duration.ofMillis(Long.MAX_VALUE)) // Infinite retention
    );

This tells Kafka Streams never to evict old window data from the state store. Repeat this pattern for your daily, weekly, and monthly windows—each gets its own dedicated store with infinite retention.

For your "permanent open window" (a cumulative aggregation without time bounds), you don't need a window at all—use a regular KTable aggregation. Configure its backing store with infinite retention the same way.

Kafka Streams creates changelog topics for each state store to replicate state across instances and recover from failures. By default, these topics have a limited retention period, so we need to override their settings to keep data forever:

Option 1: Auto-Created Topics (Set Defaults in Streams Config) #

If you let Kafka Streams auto-create changelog topics, add these settings to your streams configuration to apply to all auto-created topics:

default.topic.config={
  "retention.ms": "-1",          # Keep messages forever
  "cleanup.policy": "delete",    # Use delete policy (window keys are unique, so compaction isn't needed)
  "segment.ms": "604800000"      # Create a new segment once a week to reduce file clutter
}

The cleanup.policy=delete paired with retention.ms=-1 ensures no data is ever removed from the changelog.

Option 2: Pre-Created or Existing Topics (Modify Directly) #

If you've already created the changelog topics, use the Kafka CLI to update their configurations:

# Example for hourly changelog topic
kafka-configs.sh --bootstrap-server your-broker:9092 --alter --topic hourly-agg-store-changelog --add-config retention.ms=-1,cleanup.policy=delete,segment.ms=604800000

Repeat this for all your changelog topics (daily, weekly, monthly, and the permanent aggregation's changelog).

• Use Windowed Keys Explicitly: When working with windowed aggregations, the key in the changelog is a Windowed<K> object (combines your original key with the window's start/end time). This ensures each window's result is a unique entry in the changelog, so older windows won't be overwritten by new ones.
• Align Suppression Retention: If you're using suppress() to only emit final window results, make sure its retention period matches your window retention (infinite). For example:

  .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded().withRetention(Duration.ofMillis(Long.MAX_VALUE))))
  

  This ensures intermediate results are retained until the window closes, and the final result is kept forever.
• Name State Stores Clearly: Always use Materialized.as("store-name") to give your state stores distinct names. This makes it easier to identify and configure their corresponding changelog topics later.

• Monitor Disk Usage: Infinite retention means your changelog topics will grow indefinitely. Set up monitoring for Kafka broker disk space to avoid running out of storage.
• Scale Your Cluster: As data accumulates, ensure your Kafka cluster has enough brokers and storage capacity to handle the load. If supported, consider tiered storage to offload older data to cheaper cold storage while keeping it accessible.
• Implement Backups: Even with infinite retention, periodically back up your changelog topics to a secondary storage system (like object storage) as a safety net against data corruption or cluster failures.

内容的提问来源于stack exchange，提问作者Ajinkya Bambal