Apache Kafka

36 posts

datadog3 min readCurated summary

Achieving relentless Kafka reliability at scale with the Streaming Platform

Datadog built a Streaming Platform to make Kafka resilient and dynamically manageable across hundreds of clusters, thousands of topics, and millions of partitions. By decoupling applications from physical Kafka resources, the platform can reroute traffic, rebalance workloads, and fail over in seconds without redeployments. Its design trades strict processing order for parallelism and durability, while adding mechanisms to prevent poisoned events and backlogs from blocking entire partitions. ## A Fluid Control Plane for Multi-Cluster Orchestration - Kafka infrastructure is treated as interchangeable commodity hardware rather than a fixed application dependency. - Workloads can shift between clusters, availability zones, and Kubernetes environments automatically. - The control plane continuously replaces unhealthy components and maintains uninterrupted data flow. ### Resilient Pipelines with Streams - Streams abstract away Kafka topics and clusters behind stable identifiers. - A single Stream can span multiple Kafka clusters, availability zones, and Kubernetes clusters. - Producers and consumers do not need to know the underlying Kafka topology. - Infrastructure can be reconfigured in real time without application changes or redeployments. ### Live Traffic Failovers - When a cluster becomes unhealthy, the platform creates a replacement topic and redirects new traffic to it. - Consumers drain the backlog from the original topic before transitioning to the replacement. - The same mechanism supports proactive traffic redistribution, cluster decommissioning, capacity changes, and partition adjustments. - Operations that traditionally take hours can be completed in seconds. ### Consumer Semantics for Scale and Durability - Datadog uses at-least-once delivery rather than strict processing order. - Events can be processed independently and in parallel. - Ordering is restored later in the event store. - This trade-off enables efficient processing of petabytes of data across distributed infrastructure. ### The Assigner Coordinator - Kafka’s default group coordinator depends on session timeouts, making failover detection take tens of seconds or minutes. - Datadog’s Assigner monitors cluster health, resource usage, and workload distribution continuously. - It reacts in seconds to failures, traffic spikes, and capacity changes. - Workloads are balanced using real metrics such as CPU load and available resources, allowing heterogeneous environments to be used efficiently. ## Preventing Head-of-Line Blocking Kafka’s strict per-partition ordering means a single unprocessable event can block all later events. Datadog addresses this reliability problem through independent Stream lanes and a more flexible commit log. ### Stream Lanes and Quality of Service - Streams contain separate lanes for different priority levels and traffic requirements. - High-priority real-time traffic can be isolated from slower bursts or late-arriving data. - A dedicated dead-letter queue lane receives poison pills that cannot be processed. - Consumers can commit progress after moving failed events to the DLQ, preventing one bad event from blocking the partition without losing data. ### Advanced Commit Logging - Kafka traditionally maintains one committed pointer per partition. - That model prevents consumers from advancing past older events that are delayed or still being processed. - Datadog uses Kafka’s commit metadata to record multiple offsets or offset ranges simultaneously. - This allows consumers to continue processing live traffic while older events are handled separately, reducing backlog-related blocking. ## Overall Design Philosophy - The Streaming Platform combines stable logical Streams, real-time orchestration, flexible consumer semantics, QoS isolation, and enhanced offset tracking. - Together, these components make Kafka more self-healing and suitable for Datadog’s extreme scale. - The approach prioritizes uninterrupted processing, rapid recovery, and operational flexibility over Kafka’s default assumptions of fixed topology and strict ordering. Datadog’s design demonstrates that Kafka at very large scale requires a control layer around the broker infrastructure. Abstracting resources, automating failovers, and allowing independent progress through partitions can provide substantially better reliability and throughput than relying solely on Kafka’s native coordination model.

Read original(opens in new tab)
datadog1 min readCurated summary

Achieving relentless Kafka reliability at scale with the Streaming Platform | Datadog

The provided content does not include the blog post itself. It contains Datadog’s navigation menu and a promotional link, including a URL suggesting an article about building a Kafka streaming platform with custom abstractions, but no article text or technical sections to summarize. Please provide the full blog post content for an accurate summary.

Read original(opens in new tab)
datadog3 min readCurated summary

Timeseries indexing at scale

Datadog’s metrics volume grew 30× from 2017 to 2022, while customers began running increasingly complex queries. This growth exposed limitations in the Timeseries Index service, whose original indexing approach became a performance and maintenance bottleneck. The post introduces Datadog’s metrics architecture and explains how its indexing strategy evolved to handle large-scale workloads more reliably. ## Metrics Platform Architecture - **Intake** - Datadog Agents send data points through a load balancer to metrics intake. - Each point contains a metric name, timestamp, numerical value, and optional tags. - Tags such as `env`, `host`, and `service` provide dimensions for filtering, aggregation, and comparison. - Data is written to Kafka, allowing multiple consumers to process it for storage, indexing, analysis, and archiving. - **Storage** - The short-term storage layer has two services: - The Timeseries Database stores tuples of `<timeseries_id, timestamp, float64>`. - The Timeseries Index stores `<timeseries_id, tags>` mappings. - The custom Timeseries Index database is built on RocksDB and supports filtering and grouping during queries. - **Query Processing** - The distributed query layer contacts index nodes, retrieves intermediate results from the timeseries database, and combines them. - Filters such as `env:prod AND service:event-consumer` restrict results to matching data points. - Grouping by tags, such as `service`, produces separate timeseries for each group. - Aggregators such as `avg` combine values within each group. ## Why Timeseries Indexing Matters - Indexes prevent queries from scanning every timeseries associated with a metric, much like database indexes avoid full table scans. - Poorly designed or insufficient indexes can make queries slow and consume excessive CPU and memory. - As Datadog’s data volume and query complexity increased, the indexing system became a critical scalability concern. ## Automatically Generated Indexes - The original system generated indexes from live query behavior. - Slow or resource-intensive queries were recorded in a query log and analyzed periodically. - Index selection considered: - Query frequency - Execution time - Number of input timeseries identifiers scanned - Number of output identifiers returned - Highly selective queries—with a high input-to-output ratio—received indexes. - Obsolete indexes that no longer received queries were removed. - These indexes acted as materialized views, replacing expensive scans with efficient key-value lookups. ## Original Indexing Service Design - The service was written in Go and used embedded SQLite and RocksDB databases. - SQLite stored metadata, including: - Index definitions - Query logs - Query counts and timestamps - Input and output cardinalities - Query durations - Index definitions were read frequently, updated rarely, and cached entirely in memory. - Query logs were bulk-written in the background, keeping them out of the ingestion and query paths. - SQLite’s SQL interface made the metadata easy to inspect and modify manually. - RocksDB handled the high-volume write workload required to index trillions of events per day. Datadog’s experience shows that indexing strategies that work at smaller scale can become bottlenecks as data volume and query sophistication grow. Effective timeseries systems therefore need adaptive indexing, careful separation of query and ingestion workloads, and storage technologies suited to extremely high write rates.

Read original(opens in new tab)
datadog3 min readCurated summary

Husky: Exactly-once ingestion and multi-tenancy at scale

Husky, Datadog’s distributed, time-series-oriented event store, is optimized for large scans and aggregations rather than high-volume, low-latency point lookups. This makes exactly-once ingestion challenging, especially at Datadog’s multi-tenant scale. Datadog addresses the problem with deterministic, locality-aware routing that limits deduplication scope, improves storage efficiency, and supports autoscaling ingestion pipelines. ## Husky’s Ingestion Challenge - Husky separates storage and compute, allowing each to scale independently. - Its storage engine is designed primarily for large analytical scans and aggregations. - It is not optimized for massive numbers of low-latency point lookups, complicating duplicate detection during ingestion. - The ingestion system must guarantee that every event is stored exactly once while maintaining: - Multi-tenant scalability - Reasonable ingestion latency - Controlled infrastructure and storage costs ## Routing Events to Storage Shards - Datadog uses an upstream **Shard Router** to introduce locality into Kafka pipelines. - Events are deterministically assigned to shards based on their tenant, timestamp, and event ID. - Each tenant receives a list of shards rather than being permanently assigned to one shard. - A deterministic choice from that list distributes the tenant’s events while keeping the number of active shards as small as practical. - Downstream workers consume one or more shards and perform exactly-once ingestion into Husky. ## Benefits of Data Locality - **Simpler deduplication** - An event with the same timestamp and ID always reaches the same shard. - Deduplication only needs to occur within that shard. - Workers handle smaller sets of event IDs, making in-memory deduplication more efficient. - **Lower storage costs and better performance** - Each shard processes a relatively small set of tenants. - Husky stores each tenant in a separate table and does not mix tenants within files. - More tenants per writer produce more output files, increasing blob-storage costs and compaction work. - Restricting tenant cardinality reduces file creation and improves writer and compactor efficiency. ## Challenges in Deterministic Routing - **Changing shard assignments** - Tenant traffic can increase by one or two orders of magnitude. - Assignments may change when scaling a tenant across more shards or rebalancing traffic among existing shards. - **Distributed router consensus** - Every Shard Router node must make the same routing decision for a given event. - Inconsistent decisions could send duplicates to different shards and undermine exactly-once ingestion. - **Load balancing** - Shards must receive roughly equal traffic so downstream ingestion workers remain balanced. ## Time-Bounded Shard Placements - A simple deterministic mapping can select a shard using a hash of the event ID: ```text shard = shards[hash(event_id) % num_shards] ``` - This approach is cheap and stateless when all routers know the same shard list. - However, changing the shard list can cause the same event ID to map to a different shard, so assignment changes require additional coordination. - The article introduces **time-bounded Shard Placements** to preserve consistent routing while allowing tenant assignments to evolve, though the supplied excerpt ends before explaining the mechanism in detail. Datadog’s core recommendation is to combine deterministic, tenant-aware routing with carefully coordinated assignment changes. This narrows the scope of deduplication while reducing storage overhead and enabling balanced, scalable exactly-once ingestion.

Read original(opens in new tab)
datadog3 min readCurated summary

Introducing Husky, Datadog's third-generation event store

Datadog built Husky, a new event-storage system, after its original log architecture struggled with multi-tenant reliability, rapid platform growth, and evolving product requirements. The post explains how Datadog moved from metrics-oriented storage to event storage, introduced custom sharding and routing, and eventually recognized the need for a more flexible system. Husky emerged from these lessons about isolation, scalability, and retaining high-cardinality event data. ## From Metrics to Logs - Metrics systems store pre-aggregated tuples such as `<timeseries_id, timestamp, float64>`. - Aggregation makes metrics extremely efficient: millions of events in a second can become one compact datapoint, often requiring less than two bytes with delta-of-delta encoding. - This model is poorly suited to logs because logs must preserve individual events and their full context. - Metrics typically favor long-lived, low-cardinality dimensions such as: - Datacenter - Service - Pod name - Short-lived, high-cardinality fields such as transaction IDs and packet IDs are usually pre-aggregated or omitted. - Logs instead need to support: - Multi-kilobyte events - High-cardinality values such as UUIDs and stack traces - Arbitrary aggregations performed at query time ## Limitations of the Initial Logs System - Datadog’s first Logs architecture initially worked well but became vulnerable in a multi-tenant environment. - A single unhealthy or overloaded node could degrade service for every tenant in the cluster. - Scaling overloaded clusters could worsen the situation because nodes began streaming data to one another while already handling excessive read and write workloads. - Diagnosing and mitigating these cascading failures was difficult. ## Separating Storage from Clustering Datadog’s second architecture retained the same single-node storage engine but moved clustering responsibilities into dedicated services. - Storage nodes no longer knew about one another and behaved like independent one-node clusters. - Failures were isolated to the tenants assigned to a particular shard instead of spreading across the entire cluster. - A Shard Router: - Read events from Kafka - Reorganized them into shard-based Kafka partitions - Dynamically assigned tenants to an appropriate number of shards based on their recent five-minute data volume - Each shard was consumed by two storage-node replicas for redundancy. - A custom query engine tracked tenant-to-shard assignments, queried the relevant replicas, merged partial aggregates, and produced final results. ## Growth of the Event Platform - The new architecture substantially improved reliability and reduced operational burden. - Datadog expanded the platform beyond Logs to support products including: - Network Performance Monitoring - Real User Monitoring - Continuous Profiler - These products generated structured, multi-kilobyte events with storage and indexing requirements similar to logs. - As usage grew, new problems appeared: - A tenant producing a sudden burst of events could degrade query performance for other tenants sharing its shard. - Product teams requested longer retention for important but infrequently queried data, while still requiring it to remain immediately queryable. - The existing architecture was increasingly difficult to adapt to these isolation, scalability, and retention requirements, motivating the development of Husky. Datadog’s progression shows that event storage cannot simply reuse metrics-oriented designs. Systems must preserve event-level context, isolate tenants from one another, and support changing retention and query requirements as products and workloads evolve.

Read original(opens in new tab)
datadog3 min readCurated summary

Introducing Kafka-Kit: Tools for scaling Kafka

Datadog operates Kafka at extreme scale, ingesting trillions of data points daily and requiring petabytes of NVMe storage. To manage frequent data movement caused by scaling, recovery, and capacity changes, the company built Kafka-Kit, a set of operational tools that improve partition placement and replication control. Its primary tools, `topicmappr` and `autothrottle`, automate safer and more predictable Kafka operations. ## Kafka-Kit - Kafka-Kit addresses two major operational areas: - Data placement across brokers - Replication auto-throttling - Its main tools are: - `topicmappr`, for generating partition-to-broker mappings - `autothrottle`, for automatically controlling replication bandwidth ## Partition Placement with `topicmappr` `topicmappr` replaces Kafka’s `kafka-reassign-partitions.sh --generate` functionality while adding operational safeguards and placement controls. - Produces deterministic output: identical inputs generate the same partition map. - Supports minimal-movement broker replacement: - Failed brokers can be replaced without unnecessarily moving healthy partitions. - Partitions with complete in-sync replicas are normally left untouched. - Provides rack-aware placement using Kafka’s `broker.rack` metadata and ZooKeeper. - Supports placement based on: - Partition count - Storage size, enabling bin-packing and storage rebalancing - Allows replication factors to be increased or decreased while topics are running. - Generates clear summaries of: - Brokers being removed or added - Partition-level changes - Broker distribution before and after reassignment - Warnings and resulting partition-map files The tool is written in Go and can run from any system with access to Kafka’s ZooKeeper cluster. It requires topic names and broker IDs, then verifies that the brokers are live, sufficiently numerous, and properly distributed across configured localities. ## Replacing Failed Brokers For a failed broker, `topicmappr` can rebuild affected topics while limiting movement to the necessary partitions. - Existing replicas are preserved whenever possible. - Replacement brokers fill the gaps left by failed brokers. - The generated report makes the proposed changes visible before execution. - The example replaces broker `1002` with brokers `1003` and `1004`, showing the updated replica assignments and broker totals. ## Placement Strategies `topicmappr` offers multiple strategies for deciding where replicas should live, including `count` and tunable `storage` placement. ### Count Placement Strategy - The default strategy. - Balances leadership and the number of partitions held by each broker. - Works well when traffic is expected to be distributed evenly across partitions. - Does not require metrics data, allowing maps to be generated quickly. - Also attempts to maximize the number of distinct broker-to-broker replica relationships. - This avoids concentrating a broker’s partitions with the same small subset of peers, improving distribution across the cluster and its racks. ## Storage-Aware Placement - The storage strategy uses partition size when assigning replicas. - It supports storage bin-packing and rebalancing, which is important when brokers have uneven disk utilization. - This is particularly useful for Datadog’s large Kafka clusters, where storage capacity—not just partition count—can determine when data must be moved. Datadog’s approach demonstrates that Kafka’s flexible primitives can be extended with purpose-built tooling. For large deployments, deterministic assignments, rack awareness, minimal movement, and storage-based balancing can make scaling and failure recovery substantially safer and more predictable.

Read original(opens in new tab)