Retention policies: a practical Kafka & WebSockets guide

By Day 41, you've built real-time features with Kafka and WebSockets—but you're storing everything forever, which is burning money and slowing down your cluster. Retention policies let you decide how long Kafka keeps your events before discarding them, whether you need them for 24 hours or archive them to S3 for compliance. The right retention strategy balances cost, performance, and your business requirements.

Understanding Kafka retention basics

Kafka doesn't automatically delete messages. Without a retention policy, your disk fills up and your brokers slow to a crawl. Retention policies are topic-level configurations that tell Kafka when to clean up old segments—either by age, by size, or by keeping only the latest value per key. Think of it like deciding how long to keep transaction logs in your database: too short and you lose audit trails, too long and you're paying for wasted storage.

Every Kafka cluster has defaults (typically 7 days), but your topics need custom policies based on their role. A real-time user activity topic might expire after 1 day, while a financial audit log might stay for years.

Time-based: retention.ms

This is the most common approach. You set a time window, and Kafka deletes any segment older than that. For example, a ride-hailing platform might keep GPS coordinates for 48 hours (enough to handle disputes), then discard them. A real-time notification service might keep events for just 6 hours because consumers process them immediately.

The key insight: Kafka doesn't delete individual messages—it deletes entire segments (usually 1 GB files). So if your retention.ms is 24 hours but a segment contains messages from 23 hours ago and 1 hour ago, the whole segment stays. This matters for your calculations.

Let's configure a topic with 48-hour retention:

// Create topic with time-based retention
await adminClient.createTopics({
  topics: [
    {
      topic: 'gps-coordinates',
      numPartitions: 12,
      replicationFactor: 3,
      configEntries: [
        {
          name: 'retention.ms',
          value: '172800000' // 48 hours in milliseconds
        },
        {
          name: 'segment.ms',
          value: '3600000' // Roll segment every 1 hour for predictable cleanup
        }
      ]
    }
  ]
});

// Or update existing topic
await adminClient.alterConfigs({
  resources: [
    {
      type: ConfigResourceTypes.TOPIC,
      name: 'gps-coordinates',
      configEntries: [
        { name: 'retention.ms', value: '172800000' }
      ]
    }
  ]
});

In production, I pair retention.ms with a smaller segment.ms (the time before Kafka rolls to a new segment file). This ensures predictable cleanup timing. If you have retention.ms = 7 days but segment.ms = 30 days, you'll keep 30 days of data until the first segment older than 7 days is complete—not ideal.

Pro tip: Use segment.ms = 1 hour for high-volume topics. It creates more files but gives you granular control over when data expires.

Size-based: retention.bytes

Instead of time, you can say "keep the latest 10 GB of data per partition, then delete the rest." This is useful when message volume is unpredictable. A SaaS platform with bursty traffic might not know if a topic will hit 1 GB in 1 day or 7 days, so size-based retention guarantees bounded disk usage.

You can combine time and size: whichever limit is hit first wins. So you might say "keep data for 7 days OR 50 GB, whichever comes first."

// Size-based retention: keep latest 50 GB per partition
await adminClient.alterConfigs({
  resources: [
    {
      type: ConfigResourceTypes.TOPIC,
      name: 'user-events',
      configEntries: [
        { 
          name: 'retention.bytes', 
          value: '53687091200' // 50 GB in bytes
        },
        // Keep for 30 days OR 50 GB, whichever comes first
        { 
          name: 'retention.ms', 
          value: '2592000000' // 30 days
        }
      ]
    }
  ]
});

// Monitor actual usage
const topicMetadata = await adminClient.fetchTopicMetadata({ topics: ['user-events'] });
console.log('Partition sizes:', topicMetadata.topics[0].partitions.map(p => ({
  partition: p.partitionErrorCode,
  // Note: actual size requires broker metrics, not directly in client
})));

Size-based retention is especially useful for infrastructure topics (internal metrics, debug logs) where you just need recent data and don't care about historical trends.

Compaction for latest-state topics

Sometimes you don't want to delete messages—you want to keep only the latest value per key. This is log compaction. A perfect example: a user profile topic where messages are keyed by user_id. You might have 1 million versions of user #42's profile, but you only care about the current one. Compaction deletes old versions, dramatically shrinking the topic.

Compaction is essential for building local state stores in stream processing. When you start a new consumer group or rebuild a state store, you replay the compacted topic and immediately have the current state without replaying years of history.

// Create compacted topic for user profiles
await adminClient.createTopics({
  topics: [
    {
      topic: 'user-profiles',
      numPartitions: 12,
      replicationFactor: 3,
      configEntries: [
        {
          name: 'cleanup.policy',
          value: 'compact' // Instead of 'delete' (the default)
        },
        {
          name: 'min.compaction.lag.ms',
          value: '3600000' // Wait 1 hour after message is written before compacting
        },
        {
          name: 'delete.retention.ms',
          value: '86400000' // Keep tombstones (deletes) for 24 hours
        },
        {
          name: 'segment.ms',
          value: '604800000' // Compact every 7 days
        }
      ]
    }
  ]
});

// Produce profile updates
const profileKey = 'user-123';
await producer.send({
  topic: 'user-profiles',
  messages: [
    {
      key: profileKey,
      value: JSON.stringify({
        id: 'user-123',
        name: 'Alice',
        email: 'alice@example.com',
        subscription: 'premium',
        updated_at: Date.now()
      })
    }
  ]
});

// To delete a profile, send null value (tombstone)
await producer.send({
  topic: 'user-profiles',
  messages: [
    {
      key: profileKey,
      value: null // Tombstone: tells Kafka to delete all prior messages with this key
    }
  ]
});

In a trading platform, we used compaction for instrument metadata. Every time we updated a stock's sector or trading halt status, we sent a new message. The topic kept the latest state per symbol, and when a new process started, it could instantly load all 10,000 instruments from the compacted topic instead of replaying 6 months of updates.

Key configurations:

• cleanup.policy=compact: Enables compaction instead of time/size deletion.
• min.compaction.lag.ms: Don't compact messages written in the last N milliseconds. This ensures in-flight updates aren't prematurely discarded.
• delete.retention.ms: Keep tombstones (null messages) for this long so lagging consumers see the deletions.

Archive to S3 for long-term history

Kafka isn't a data warehouse. If you need to analyze events from 2 years ago, or comply with regulations requiring 7-year audit trails, you can't rely on Kafka retention. Instead, stream events to S3 using a connector or custom application, compress them, and query with tools like Athena or BigQuery.

This is how we handled a fintech platform's trade settlement records: Kafka kept 30 days of data for operational use, but a Kafka Connect S3 sink archived every trade to S3 in Parquet format, partitioned by date. Regulators could query the archive; operators could query Kafka.

// Example: simple Node.js archiver
// (In production, use Kafka Connect S3 sink for reliability)

import { Kafka } from 'kafkajs';
import { S3Client, PutObjectCommand } from '@aws-sdk/client-s3';
import { gzip } from 'zlib';
import { promisify } from 'util';

const gzipAsync = promisify(gzip);
const s3 = new S3Client({ region: 'us-east-1' });

const kafka = new Kafka({
  clientId: 'archiver',
  brokers: ['kafka-1:9092', 'kafka-2:9092']
});

const consumer = kafka.consumer({ groupId: 'archive-group' });

async function archiveToS3() {
  await consumer.connect();
  await consumer.subscribe({ topic: 'trades', fromBeginning: false });

  let batch: any[] = [];
  const batchSize = 10000;

  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      batch.push({
        key: message.key?.toString(),
        value: JSON.parse(message.value?.toString() || '{}'),
        timestamp: message.timestamp
      });

      if (batch.length >= batchSize) {
        await flushBatch(batch);
        batch = [];
      }
    }
  });
}

async function flushBatch(messages: any[]) {
  const now = new Date();
  const datePartition = `${now.getFullYear()}/${String(now.getMonth() + 1).padStart(2, '0')}/${String(now.getDate()).padStart(2, '0')}`;
  const hour = String(now.getHours()).padStart(2, '0');
  const s3Key = `kafka-archive/trades/${datePartition}/hour=${hour}/${Date.now()}.jsonl.gz`;

  const jsonl = messages.map(m => JSON.stringify(m)).join('\n');
  const compressed = await gzipAsync(jsonl);

  await s3.send(new PutObjectCommand({
    Bucket: 'my-data-lake',
    Key: s3Key,
    Body: compressed,
    ContentEncoding: 'gzip',
    ContentType: 'application/jsonl'
  }));

  console.log(`Archived ${messages.length} messages to ${s3Key}`);
}

archiveToS3();

The key benefit: your Kafka cluster stays lean (30 days of data), but you have a complete audit trail in S3. You can run Athena queries on the archive without touching Kafka.

Common mistakes

• Setting retention.ms too high without calculating disk needs. A topic with 100k msgs/sec and 30-day retention needs ~260 GB per partition. Multiply by replication factor and partition count—your disk is full before you know it. Use a calculator: (msgs/sec × seconds) × average_msg_size × replication_factor.
• Using compaction without understanding the lag. If you set min.compaction.lag.ms to 1 second but have a 10-second processing delay, old versions get deleted before consumers see them. Use 1-2 hours as a safe default.
• Mixing cleanup policies. You can't use both delete and compact on the same topic. Pick one strategy per topic.
• Forgetting to archive before setting short retention. If you delete Kafka data without backing it up first, it's gone forever. Always archive before shortening retention.
• Not monitoring actual disk usage. Set a retention policy, then check your broker metrics weekly. Topics grow slower or faster than expected. Use JMX metrics like kafka.log:type=LogCleaner,name=max-clean-time-secs and disk usage per broker.

Key takeaways

• Time-based retention (retention.ms) is the default—use it for operational topics that need recent data but not historical archives.
• Size-based retention (retention.bytes) is better for unpredictable workloads where you want bounded disk usage regardless of message rate.
• Log compaction (cleanup.policy=compact) keeps only the latest value per key, essential for state topics like user profiles, configs, or entity data.
• Pair retention.ms with segment.ms to ensure predictable cleanup; don't let segments grow larger than your retention window.
• For compliance or analytics, archive to S3 (using Kafka Connect or custom code), then set short Kafka retention to keep your cluster fast and cheap.
• Always calculate disk requirements before deploying a retention policy: (msgs/sec × retention_seconds × avg_msg_size × replication_factor × partition_count).
• Monitor actual disk usage and adjust retention quarterly as traffic patterns shift.