Summary
What this post covers: An end-to-end blueprint for building a production-grade Kafka ingestion engine for multivariate server time series, including psutil collection, Avro schema design, a tuned Python producer, partitioning, retention, and downstream consumer patterns.
Key insights:
- Kafka belongs between collectors and storage because it decouples failure modes—when InfluxDB or TimescaleDB goes down, producers keep writing and consumers replay from the log rather than dropping data.
- Correlated multivariate metrics should be emitted as a single Avro record on one topic; splitting them across topics forces consumers to perform expensive joins and defeats the purpose of capturing them together.
- Partition by hostname or instance ID—never by timestamp, since monotonic timestamps create rolling hot spots—and keep partition count comfortably larger than host count for even load distribution.
- Tuning
linger.ms,batch.size, andcompression.type(lz4 or snappy) can raise a single Python producer’s throughput by more than an order of magnitude over the conservative defaults, while keeping tail latency bounded. - Set Schema Registry compatibility to
BACKWARDand give every new Avro field a default value, then deploy schema → producer → consumer in that order to evolve safely without breaking running consumers.
Main topics: The Challenge of Server Telemetry at Scale, Why Kafka for Multivariate Time Series, What Multivariate Time Series Actually Means, Architecture of the Engine, Collecting Server Metrics with psutil, Designing the Avro Message Schema, Building the Kafka Producer, Partitioning Strategy for Time Series, Topic Design and Retention, Consumer Patterns and Downstream Sinks, Production Concerns, Producer Tuning and Throughput.
The Challenge of Server Telemetry at Scale
A single modern server, when fully instrumented, can easily emit more than 10,000 metric samples per second. Multiplied across a few hundred machines in a modest production fleet, this produces millions of timestamped numbers per second — all correlated, all required, and all useless if they cannot be stored and replayed reliably. This is where most homegrown monitoring stacks quietly fail. A script that scrapes /proc/stat every five seconds and pushes rows directly into a time-series database appears elegant in a demonstration, but the moment the database goes down for maintenance, the collector crashes, or a network disruption drops packets, the data is lost permanently. For observability, missing data is often more harmful than no data at all, because dashboards continue drawing lines and the gap goes unnoticed.
A representative incident illustrates the point: a fleet of ingestion machines began dropping metrics during a peak load spike. Grafana dashboards interpolated across the gap, and three full days passed before anyone recognised that the next quarter’s capacity plan had been built on fabricated values. That incident underscores a principle that has guided every observability pipeline since: the boundary between systems that produce data and systems that store data is one of the most important boundaries in a distributed architecture, and Apache Kafka remains the most appropriate component to sit on that boundary.
This guide describes how to build a production-grade Kafka time-series engine end to end. The discussion covers collecting multivariate metrics from a Linux server, serializing them with Avro, pushing them through a tuned Python producer, routing them through deliberate partitioning, and feeding them to downstream consumers that depend on them. Working code, complete Avro schemas, copy-ready configuration, and the kind of detail that surfaces only after observing production failures are all provided.
Why Kafka for Multivariate Time Series
Before any code is written, the question every engineer raises when Kafka is proposed deserves a direct answer: is Kafka actually required? Kafka is not the cheapest or simplest tool in the observability toolbox, but it is almost always the appropriate one once a deployment outgrows a single machine and a single storage target. Five properties make Kafka indispensable for multivariate time series, and each one addresses a specific failure mode that emerges the first time a stack of this kind is built without Kafka in the middle.
The first property is durability. Kafka persists every message to disk before acknowledging it, and with replication factor three, two broker failures can be tolerated without data loss. Time-series databases such as InfluxDB or TimescaleDB are durable in their own right, but they are stateful, tuned for query performance, and frequently the first systems taken down during an upgrade. When producers write directly to the database, an upgrade window becomes a data-loss window. With Kafka in the middle, producers continue writing, Kafka continues storing, and the database catches up when it returns.
The second property is replay. Because Kafka retains data for a configurable window — hours, days, or weeks — any consumer can reset its offset and re-read history. This transforms incident postmortems from inference based on prior dashboards into precise replay of the exact data the monitoring system observed. It is also how a new downstream system is onboarded: a fresh consumer is pointed at earliest and catches up.
The third property is fan-out. Metrics are rarely consumed by a single system. A typical deployment includes a long-term store, a fast-access store, a stream processor for alerting, and possibly a machine-learning training sink. Kafka allows any number of independent consumer groups to attach to the same topic without coordination between them. Each group reads at its own pace, and a slow consumer cannot apply back-pressure to a fast one.
The fourth property is decoupling. The producer requires no knowledge of the consumer, and vice versa. InfluxDB can be swapped for TimescaleDB without modifying a single line of collector code. This is the same argument that motivated the move toward microservices, and it applies with equal force to data pipelines. For an examination of this decoupling at the storage layer, the time series database comparison guide reviews the tradeoffs between common sinks.
The fifth property is horizontal scale. A single Kafka topic can be partitioned across dozens or hundreds of brokers, and each partition is an independent log. As a fleet grows from fifty servers to five thousand, partitions and brokers are added rather than the pipeline being rewritten. The same Kafka cluster architecture can scale by orders of magnitude in throughput without a fundamental redesign, which is not a property most alternatives can claim.
What Multivariate Time Series Actually Means
The term “multivariate time series” is often used loosely, so a precise definition is in order. A univariate time series is a single signal indexed by time — for example, CPU utilisation sampled every second. A multivariate time series is a collection of two or more signals sampled at the same timestamps that are correlated with one another. On a server, CPU rarely matters in isolation. It matters together with memory pressure, disk I/O wait, network throughput, and possibly temperature, because the meaningful patterns reside in the relationships between those signals.
Consider a representative example: a sudden CPU spike. In isolation, it conveys little information. If, at the same timestamp, memory usage is also climbing, disk I/O is dropping to near zero, and network bytes per second are flatlining, the signature most likely indicates a CPU-bound computation, perhaps a runaway regular expression or a JVM in a garbage-collection storm. By contrast, a CPU spike accompanied by high iowait, growing disk queue depth, and falling network throughput indicates disk saturation causing downstream throttling. These diagnoses are possible only because the signals arrive together, on the same timeline, in the same record.
This has two concrete implications for engine design. First, all signals should be captured at the same instant in a single message rather than as separate messages per metric. Second, the storage and query layer should make it inexpensive to align those signals on the time axis, which is precisely what purpose-built time-series databases provide. For a deeper treatment of forecasting on this type of data, the guide on time series forecasting models describes how models exploit the correlations captured here.
The chart above illustrates how CPU and memory climb together during the middle of the window while disk I/O and network activity move in the opposite direction. This divergence is the principal reason for capturing these signals together. Storing them in different Kafka topics with different timestamps and different partitioning schemes results in downstream query effort being dominated by realignment, which should be avoided.
Architecture of the Engine
The engine has four layers, and the most useful way to conceptualise them is as a sequence in which each layer must hand off correctly to the next.
Layer one is collection. On each server, a small Python process samples metrics at a fixed interval (typically one second) using psutil. It bundles CPU, memory, disk, and network counters into a single record keyed by hostname and timestamp. This process runs as a systemd service and consumes minimal resources, with negligible steady-state CPU on a small instance.
Layer two is production. The same Python process serializes each record using an Avro schema fetched from the Schema Registry, then hands it to a confluent-kafka-python producer configured for durability and throughput. The producer batches records, compresses them with lz4, and sends them to the broker with acks=all.
Layer three is the broker. Kafka persists the records to a topic called server.metrics.v1, partitioned by hostname. Replication factor three ensures no data loss on broker failure. The topic has a retention of 72 hours, which is sufficient to replay into a new consumer without exhausting broker disk.
Layer four is consumption. Multiple independent consumer groups read from the topic. One writes to InfluxDB for long-term storage, one runs Flink jobs for windowed aggregations and anomaly detection, and one feeds a lightweight alerting service. Each may be deployed, restarted, or replaced without affecting the others. For a local Kafka environment suitable for development, the Docker containers guide covers the necessary container basics.
Collecting Server Metrics with psutil
The psutil library is the appropriate tool for cross-platform metric collection in Python. It provides CPU, memory, disk, and network statistics through a consistent interface that operates identically on Linux, macOS, and Windows. One rule must be observed: many of its counters are cumulative — for example, psutil.net_io_counters() returns total bytes since boot rather than bytes per second — so a delta between two consecutive samples is required to derive a rate.
The following is a clean collector that captures a multivariate sample at each tick:
import socket
import time
from dataclasses import dataclass, asdict
from typing import Optional
import psutil
@dataclass
class MetricSample:
host: str
timestamp_ms: int
cpu_percent: float
cpu_user: float
cpu_system: float
cpu_iowait: float
mem_percent: float
mem_used_bytes: int
mem_available_bytes: int
swap_percent: float
disk_read_bytes_per_sec: float
disk_write_bytes_per_sec: float
disk_read_iops: float
disk_write_iops: float
net_rx_bytes_per_sec: float
net_tx_bytes_per_sec: float
net_rx_packets_per_sec: float
net_tx_packets_per_sec: float
load_1m: float
load_5m: float
load_15m: float
class MetricCollector:
def __init__(self, interval_seconds: float = 1.0):
self.interval = interval_seconds
self.host = socket.gethostname()
self._prev_disk = psutil.disk_io_counters()
self._prev_net = psutil.net_io_counters()
self._prev_time = time.monotonic()
# First CPU call is non-blocking and returns 0.0; prime it.
psutil.cpu_percent(interval=None)
psutil.cpu_times_percent(interval=None)
def sample(self) -> MetricSample:
now = time.monotonic()
elapsed = max(now - self._prev_time, 1e-6)
cpu_pct = psutil.cpu_percent(interval=None)
cpu_times = psutil.cpu_times_percent(interval=None)
vm = psutil.virtual_memory()
sm = psutil.swap_memory()
load = psutil.getloadavg()
disk = psutil.disk_io_counters()
d_read_b = (disk.read_bytes - self._prev_disk.read_bytes) / elapsed
d_write_b = (disk.write_bytes - self._prev_disk.write_bytes) / elapsed
d_read_iops = (disk.read_count - self._prev_disk.read_count) / elapsed
d_write_iops = (disk.write_count - self._prev_disk.write_count) / elapsed
net = psutil.net_io_counters()
n_rx_b = (net.bytes_recv - self._prev_net.bytes_recv) / elapsed
n_tx_b = (net.bytes_sent - self._prev_net.bytes_sent) / elapsed
n_rx_p = (net.packets_recv - self._prev_net.packets_recv) / elapsed
n_tx_p = (net.packets_sent - self._prev_net.packets_sent) / elapsed
self._prev_disk = disk
self._prev_net = net
self._prev_time = now
return MetricSample(
host=self.host,
timestamp_ms=int(time.time() * 1000),
cpu_percent=cpu_pct,
cpu_user=cpu_times.user,
cpu_system=cpu_times.system,
cpu_iowait=getattr(cpu_times, "iowait", 0.0),
mem_percent=vm.percent,
mem_used_bytes=vm.used,
mem_available_bytes=vm.available,
swap_percent=sm.percent,
disk_read_bytes_per_sec=d_read_b,
disk_write_bytes_per_sec=d_write_b,
disk_read_iops=d_read_iops,
disk_write_iops=d_write_iops,
net_rx_bytes_per_sec=n_rx_b,
net_tx_bytes_per_sec=n_tx_b,
net_rx_packets_per_sec=n_rx_p,
net_tx_packets_per_sec=n_tx_p,
load_1m=load[0],
load_5m=load[1],
load_15m=load[2],
)
Several details are worth highlighting. The implementation uses time.monotonic() for the elapsed calculation because it is immune to wall-clock adjustments; if NTP shifts the system clock backwards, time.time() deltas can become negative and produce meaningless rates. time.time() is still used for the sample timestamp itself because downstream consumers expect wall-clock time. getattr is used for iowait because it exists only on Linux; on macOS it silently returns zero.
Regarding the hostname: augmenting it with cloud metadata (instance ID, region, availability zone) is strongly recommended when running on AWS, GCP, or Azure. Hostnames are acceptable as a partition key but can collide across environments, and during incident triage it is essential to identify the exact instance that emitted an anomalous value. The related article on managing metadata for time series signals describes this pattern in greater detail.
Designing the Avro Message Schema
Every production Kafka deployment eventually suffers from the absence of a schema, typically on the day another team adds a new field to the producer and the downstream consumer begins throwing KeyError exceptions in the middle of the night. Avro with a Schema Registry addresses this by making the schema a first-class part of the message itself. Producers register their schema once, and every message carries a five-byte prefix with the schema ID. Consumers use that ID to fetch the exact schema the producer used and deserialize deterministically. It is one of the most valuable components in the Kafka ecosystem, and it can be set up in approximately fifty lines of code.
The following is the Avro schema for the multivariate sample. It should be saved as schemas/server_metric.avsc:
{
"type": "record",
"name": "ServerMetric",
"namespace": "com.aicodeinvest.metrics",
"doc": "A multivariate sample of host-level server metrics.",
"fields": [
{"name": "host", "type": "string", "doc": "Hostname or instance ID"},
{"name": "timestamp_ms", "type": "long", "doc": "Unix epoch ms"},
{"name": "cpu_percent", "type": "double"},
{"name": "cpu_user", "type": "double"},
{"name": "cpu_system", "type": "double"},
{"name": "cpu_iowait", "type": "double", "default": 0.0},
{"name": "mem_percent", "type": "double"},
{"name": "mem_used_bytes", "type": "long"},
{"name": "mem_available_bytes", "type": "long"},
{"name": "swap_percent", "type": "double", "default": 0.0},
{"name": "disk_read_bytes_per_sec", "type": "double"},
{"name": "disk_write_bytes_per_sec", "type": "double"},
{"name": "disk_read_iops", "type": "double"},
{"name": "disk_write_iops", "type": "double"},
{"name": "net_rx_bytes_per_sec", "type": "double"},
{"name": "net_tx_bytes_per_sec", "type": "double"},
{"name": "net_rx_packets_per_sec", "type": "double"},
{"name": "net_tx_packets_per_sec", "type": "double"},
{"name": "load_1m", "type": "double"},
{"name": "load_5m", "type": "double"},
{"name": "load_15m", "type": "double"},
{"name": "tags", "type": {"type": "map", "values": "string"}, "default": {}}
]
}
Three design decisions merit explanation. First, every field that is not strictly required carries a default. This makes schema evolution safe: if gpu_percent is added with a default of zero, older consumers unaware of GPUs can still deserialize new messages without crashing. The Schema Registry enforces this rule automatically when the compatibility mode is set to BACKWARD, which is the recommended configuration.
Second, a free-form tags map is included. Tags hold values such as environment, region, team, and cluster ID — anything that varies between deployments and may be useful for downstream filtering. Keeping them in a map rather than as top-level fields permits new tags to be added without a schema change. A small serialization cost is incurred, but it is negligible compared with the operational overhead of coordinating schema updates.
Third, nested records are avoided. Avro supports them, but flat schemas serialize faster, are easier to query in downstream SQL systems, and integrate more smoothly with Kafka Connect sinks. For metrics specifically, a flat schema is almost always the appropriate choice.
BACKWARD means new consumers can read old messages, FORWARD means old consumers can read new messages, and FULL means both. For metrics, BACKWARD is usually sufficient, but the team should agree on the mode before the first producer is deployed. Changing the compatibility mode on a running topic is operationally painful.
Building the Kafka Producer
The collector and the schema are now combined into a working producer. The implementation uses confluent-kafka-python, which wraps the production-proven librdkafka C library and is significantly faster than the pure-Python alternatives. For readers interested in the performance gap between Python and compiled languages on this kind of workload, the Python vs Rust comparison guide provides context, but for metric producers Python is almost always sufficient when the appropriate client is used.
import json
import logging
import signal
import sys
import time
from dataclasses import asdict
from confluent_kafka import Producer, KafkaError
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
from confluent_kafka.serialization import (
SerializationContext,
MessageField,
StringSerializer,
)
from collector import MetricCollector, MetricSample
log = logging.getLogger("kafka-metrics")
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
TOPIC = "server.metrics.v1"
def load_schema(path: str) -> str:
with open(path) as f:
return f.read()
def to_dict(sample: MetricSample, ctx) -> dict:
return asdict(sample)
def delivery_report(err, msg):
if err is not None:
log.error("delivery failed for key=%s: %s", msg.key(), err)
# Success path is intentionally silent — we would drown in logs otherwise.
def build_producer() -> Producer:
conf = {
"bootstrap.servers": "kafka-1:9092,kafka-2:9092,kafka-3:9092",
"client.id": "metric-collector",
# Durability
"acks": "all",
"enable.idempotence": True,
"max.in.flight.requests.per.connection": 5,
"retries": 10_000_000,
"delivery.timeout.ms": 120_000,
# Throughput
"linger.ms": 20,
"batch.size": 65_536,
"compression.type": "lz4",
# Memory bound
"queue.buffering.max.messages": 100_000,
"queue.buffering.max.kbytes": 1_048_576,
}
return Producer(conf)
def main():
sr_client = SchemaRegistryClient({"url": "http://schema-registry:8081"})
avro_serializer = AvroSerializer(
schema_registry_client=sr_client,
schema_str=load_schema("schemas/server_metric.avsc"),
to_dict=to_dict,
)
key_serializer = StringSerializer("utf_8")
producer = build_producer()
collector = MetricCollector(interval_seconds=1.0)
running = True
def shutdown(signum, frame):
nonlocal running
log.info("shutdown requested, flushing producer...")
running = False
signal.signal(signal.SIGTERM, shutdown)
signal.signal(signal.SIGINT, shutdown)
next_tick = time.monotonic()
try:
while running:
sample = collector.sample()
key = key_serializer(sample.host)
value = avro_serializer(
sample,
SerializationContext(TOPIC, MessageField.VALUE),
)
producer.produce(
topic=TOPIC,
key=key,
value=value,
timestamp=sample.timestamp_ms,
on_delivery=delivery_report,
)
# Serve delivery callbacks without blocking.
producer.poll(0)
next_tick += collector.interval
sleep_for = next_tick - time.monotonic()
if sleep_for > 0:
time.sleep(sleep_for)
else:
# Fell behind; log once and resync.
log.warning("collector is behind by %.3fs", -sleep_for)
next_tick = time.monotonic()
finally:
remaining = producer.flush(timeout=30)
if remaining > 0:
log.error("%d messages undelivered at shutdown", remaining)
sys.exit(1)
log.info("clean shutdown")
if __name__ == "__main__":
main()
Each of the configuration choices warrants explanation because each contributes specific behaviour.
acks=all instructs the broker to wait until all in-sync replicas have written the message before acknowledging. Combined with enable.idempotence=true, this provides exactly-once semantics at the producer level: retries will not duplicate messages even if the network drops an acknowledgment. This is the single most important configuration for durability and should not be disabled outside of throwaway demonstrations.
linger.ms=20 instructs the producer to wait up to twenty milliseconds before sending a batch, even when the batch is not full. This represents a throughput-versus-latency tradeoff. For metrics sampled at 1 Hz, the additional latency is negligible, while throughput can increase by a factor of five to ten because network and serialization overhead is amortised across many records.
batch.size=65536 sets the maximum size of a single batch. With twenty milliseconds of linger and a reasonable message rate, each batch typically fills before the timer fires.
compression.type=lz4 is the recommended default for metrics. It compresses repetitive numeric data well (often by a factor of three to five) and is faster than both snappy and zstd at reasonable compression levels. Benchmarking against actual data is advisable, but lz4 rarely underperforms.
The table below summarises how these configuration choices trade off, together with common alternatives:
| Setting | Value | Tradeoff |
|---|---|---|
acks |
all | Durability over latency. Worth every millisecond. |
enable.idempotence |
true | Exactly-once producer semantics. No duplicates on retry. |
linger.ms |
20 | Up to 20ms extra latency for 5–10x throughput. |
compression.type |
lz4 | Fastest high-ratio compression for numeric data. |
batch.size |
65,536 | Large batches amortize network costs. |
max.in.flight |
5 | Max allowed with idempotence. Higher values are rejected. |
Partitioning Strategy for Time Series
Selecting the wrong partition key is the most common and most damaging mistake in a Kafka time-series deployment. The challenge is that partitioning has two competing goals: records from the same logical entity should land on the same partition so that their order is preserved, while load should be distributed evenly across partitions so that no single partition becomes a hotspot. For time series, one tempting choice is to use the timestamp. The timestamp should never be used as a partition key. A monotonic timestamp creates a pathological pattern in which every new record lands on whichever partition is currently hottest, producing a rolling hotspot that shifts across partitions over time.
The partition keys that work well for multivariate server metrics are all variations on the same principle: key by the source of the data. The main options are:
| Strategy | Good for | Watch out for |
|---|---|---|
| hostname | Most fleets. Preserves per-host ordering. | Imbalance if one host is much busier. |
| cluster_id + hostname | Multi-tenant setups where clusters are the billing unit. | Cluster-sized hot spots. |
| metric_family | When consumers only care about one family. | Small number of partitions—only as many as families. |
| random/sticky | Perfectly even load, no ordering needs. | Loses per-host ordering. |
| timestamp | Never. | Rolling hot spots, reprocessing nightmares. |
For nearly every deployment encountered in practice, partitioning by hostname is the correct default. It preserves per-host ordering (which matters because consumers often perform stateful work per host, such as anomaly detection), and it distributes load evenly as long as the partition count is comfortably larger than the host count. The modern Kafka client defaults to the sticky partitioner for records without a key, which is a useful throughput optimisation; since a key is being provided here, that optimisation does not apply, and records are routed to hash(hostname) % partition_count.
One recommendation is particularly important: set the partition count to a round number that is comfortably larger than the current fleet and grows in increments of five or ten — for example, thirty, fifty, or one hundred, rather than twenty-three or forty-seven. Kafka supports adding partitions to a topic, but doing so changes the hash mapping for keyed records, which is a substantive operational disruption. Begin with headroom.
Topic Design and Retention
The question of whether to use one topic for all metrics or a topic per metric family arises frequently. The answer for multivariate time series is almost always one topic. The very purpose of capturing correlated signals together is that downstream consumers require them together. Splitting them into separate topics forces every consumer to join across topics to reconstruct a sample, which is precisely the complexity Kafka is intended to mitigate.
Exceptions are rare but real. When fundamentally different data types have different retention or sizing requirements — for example, high-frequency metrics and low-frequency events — placing them in separate topics is reasonable, because different retention policies typically apply. Within “host metrics” itself, however, one topic is the right answer.
The following is a reasonable topic configuration for a production multivariate metrics topic, applied with kafka-topics.sh:
kafka-topics.sh --bootstrap-server kafka-1:9092 \
--create \
--topic server.metrics.v1 \
--partitions 50 \
--replication-factor 3 \
--config retention.ms=259200000 \
--config segment.bytes=536870912 \
--config compression.type=producer \
--config min.insync.replicas=2 \
--config cleanup.policy=delete \
--config max.message.bytes=1048576
The significant settings are as follows: retention.ms=259200000 retains data for three days, which is sufficient to reprocess into a new sink or recover from a downstream outage without exhausting broker disks. segment.bytes=536870912 (512 MiB) controls when a new log segment is rolled; larger segments mean fewer files and faster startup but coarser cleanup granularity. compression.type=producer instructs the broker to store messages in whatever format the producer sent, avoiding unnecessary decompress-recompress cycles. min.insync.replicas=2 combined with acks=all on the producer is what actually provides durability; acks=all alone offers no guarantee if only one replica is in sync.
Finally, cleanup.policy=delete is almost always appropriate for metrics. Log compaction (the alternative) retains the latest record per key, which is suitable for changelog streams but inappropriate for time series, where every record matters.
Consumer Patterns and Downstream Sinks
Once data is in Kafka, consumers are comparatively straightforward. The following is a minimal consumer that reads multivariate samples and writes them to InfluxDB. For an end-to-end treatment of that pipeline, the article on InfluxDB to Iceberg with Telegraf covers the long-term storage side in depth.
from confluent_kafka import Consumer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroDeserializer
from confluent_kafka.serialization import SerializationContext, MessageField
from influxdb_client import InfluxDBClient, Point, WriteOptions
TOPIC = "server.metrics.v1"
sr_client = SchemaRegistryClient({"url": "http://schema-registry:8081"})
avro_deser = AvroDeserializer(schema_registry_client=sr_client)
consumer = Consumer({
"bootstrap.servers": "kafka-1:9092",
"group.id": "influxdb-sink",
"auto.offset.reset": "latest",
"enable.auto.commit": False,
"max.poll.interval.ms": 300_000,
"session.timeout.ms": 30_000,
})
consumer.subscribe([TOPIC])
influx = InfluxDBClient(url="http://influxdb:8086", token="...", org="aic")
write_api = influx.write_api(write_options=WriteOptions(batch_size=5_000, flush_interval=2_000))
try:
while True:
msg = consumer.poll(1.0)
if msg is None:
continue
if msg.error():
print(f"consumer error: {msg.error()}")
continue
record = avro_deser(
msg.value(),
SerializationContext(msg.topic(), MessageField.VALUE),
)
point = (
Point("server_metrics")
.tag("host", record["host"])
.field("cpu_percent", record["cpu_percent"])
.field("mem_percent", record["mem_percent"])
.field("disk_read_bps", record["disk_read_bytes_per_sec"])
.field("disk_write_bps", record["disk_write_bytes_per_sec"])
.field("net_rx_bps", record["net_rx_bytes_per_sec"])
.field("net_tx_bps", record["net_tx_bytes_per_sec"])
.field("load_1m", record["load_1m"])
.time(record["timestamp_ms"], "ms")
)
write_api.write(bucket="metrics", record=point)
consumer.commit(msg, asynchronous=True)
finally:
consumer.close()
write_api.close()
influx.close()
Several consumer-side details matter. Auto-commit is disabled because commits should be tied to successful downstream writes; the pattern is “write, then commit the offset that was just written,” which provides at-least-once semantics end to end. The InfluxDB write API is used with batching for the same reason batching is used on the producer: per-record writes are slow, while batches are fast.
For more sophisticated consumers — particularly anything that requires windowing, joins, or complex event patterns — a plain consumer should be replaced by a full stream processor. Flink CEP is a common choice; the Flink CEP pipeline guide describes precisely the kind of pattern that can be built on top of this Kafka topic.
Production Concerns
Everything described above works in a demonstration. To run it in production, five additional concerns must be addressed: monitoring consumer lag, handling backpressure at the producer, handling broker failures, managing exactly-once semantics, and capacity planning.
Consumer lag is the single most important metric to monitor on this pipeline. It indicates whether consumers are keeping pace with producers. The standard tool is kafka-consumer-groups.sh, but continuous monitoring is better served by Kafka’s built-in JMX metrics or a tool such as Burrow or Kafka Exporter feeding Prometheus. Alerts should fire on sustained lag growth rather than absolute lag values; a transient bump during a deployment is normal, while lag that has been growing for five minutes is a problem.
Backpressure at the producer appears as a full internal queue. In confluent-kafka-python, producer.produce() raises BufferError when the queue is full. Two responses are possible: block until space becomes available (which eventually blocks the metric collector), or drop samples (which produces gaps but keeps the collector responsive). For metrics, the first option, bounded by a timeout, is usually preferable, because dropped samples can conceal incidents. The pattern is as follows:
from confluent_kafka import KafkaException
def produce_with_backpressure(producer, topic, key, value, ts):
for attempt in range(3):
try:
producer.produce(
topic=topic, key=key, value=value, timestamp=ts,
on_delivery=delivery_report,
)
return
except BufferError:
# Internal queue is full; poll to serve callbacks and drain.
producer.poll(0.5)
log.error("dropping sample for %s after 3 backpressure retries", key)
Broker failures are handled automatically by the client when the configuration is correct. With acks=all, enable.idempotence=true, and retries set to an effectively unbounded value, a broker outage causes the producer to retain messages in its buffer and retry until a new leader is elected. The delivery.timeout.ms setting is the ultimate deadline; messages older than that are considered failed and returned through the delivery callback.
Exactly-once semantics is overloaded terminology. The producer provides exactly-once delivery to the broker through idempotence. End-to-end exactly-once from producer to downstream sink requires the sink to be idempotent as well — either because it is naturally idempotent (upserts, deduplication by key plus timestamp) or because it participates in Kafka transactions. For metrics, full transactions are rarely required; at-least-once plus an idempotent sink (the InfluxDB write API is one such sink) is usually sufficient, because writing the same point twice merely overwrites it with the same value.
Producer Tuning and Throughput
The most important lever on producer throughput is how records are batched and compressed. The table below describes, qualitatively, how each successive configuration change affects throughput and latency for a single Python producer sending small (a few hundred bytes) metric records. Absolute numbers depend heavily on hardware, broker count, record size, and partition count, so the table shows the direction and relative magnitude of each change rather than figures from any single run.
| Configuration | Relative throughput | Typical latency | Tail latency |
|---|---|---|---|
| No batching, no compression | Lowest (baseline) | Lowest | Lowest |
| linger.ms=5, snappy | Substantially higher | Slightly higher | Comparable |
| linger.ms=20, lz4 | Higher still | Moderately higher | Moderately higher |
| linger.ms=50, lz4, 128KB batches | Highest | Highest | Highest, still bounded |
Several observations follow. First, batching and compression together improve throughput by more than an order of magnitude over the naive configuration. Second, the latency cost is real but modest; even at the most aggressive setting, tail latency remains within a range that is acceptable for metrics. Third, a single well-tuned Python producer on modest hardware can sustain a high enough rate that one producer comfortably handles a fleet of hundreds or thousands of hosts at 1 Hz sampling. Running one producer per server is not required when aggregation is preferred.
Compression ratios on metric data also merit attention. Repetitive numeric records compress well under lz4 — commonly by a factor of three to five — which reduces network and broker disk cost proportionally. In a large fleet this represents one of the largest savings in the entire pipeline.
confluent-kafka-python are conservative. Setting linger.ms, batch.size, and compression.type can improve producer throughput by more than an order of magnitude over the defaults. These three settings should be tuned first, with all other adjustments following.
Frequently Asked Questions
Why Kafka instead of writing directly to InfluxDB or TimescaleDB?
Direct-to-database works until something breaks. When the database is down for maintenance, your collector crashes or backs up. When you want to add a second consumer—say, an alerting service—you either double-write from the collector (error-prone) or read back from the database (slow and fragile). Kafka puts a durable, replayable buffer between producers and consumers, which decouples the failure modes of the two sides. For a small single-sink deployment, direct writes are fine. For anything where observability matters during incidents, Kafka is worth the extra moving part.
How many messages per second can a single Python producer handle?
With the configuration in this post (linger.ms=20, lz4 compression, and larger batches), a single well-tuned confluent-kafka-python producer on modest hardware sustains throughput more than an order of magnitude above the conservative defaults, which is comfortably more than enough for a fleet of thousands of hosts at 1Hz sampling. If more is needed, the usual answer is not a faster producer, it is multiple producers, one per host or one per small group of hosts, which also gives better fault isolation.
Should I use one topic or multiple topics for different metric types?
For multivariate metrics that are correlated and consumed together, use one topic. Splitting them into separate topics forces downstream consumers to join across topics, which defeats the purpose of capturing multivariate data in the first place. Use separate topics only when the data has genuinely different retention, sizing, or consumer profiles—for example, high-frequency metrics versus low-frequency events, or metrics versus logs.
How do I handle schema evolution when adding new metrics?
Set your Schema Registry compatibility mode to BACKWARD. When adding a field, give it a default value in the Avro schema. This lets new consumers read old messages (with the default filled in) and lets old consumers safely ignore the new field. Deploy the schema change to the registry first, then deploy the producer change, then deploy the consumer change—in that order. Never remove a field without first making sure no active consumer reads it.
What partitioning key should I use for multivariate time series?
Partition by hostname (or instance ID) in almost every case. This preserves per-host ordering, which is what stateful consumers like anomaly detectors need, and it distributes load evenly as long as your partition count is comfortably larger than your host count. Never use the timestamp as a partition key, monotonic timestamps create rolling hot spots where each new batch of records lands on the same partition.
Related Reading
Concluding Observations
Building a Kafka-based engine for multivariate time series is one of those projects that appears excessive on day one and proves foundational by month three. The core ideas are straightforward: collect correlated signals together, serialize them with a schema, partition by source, tune the producer for throughput, and allow Kafka to serve as the durable spine that decouples collectors from consumers. Everything else — the choice of time-series database, the streaming framework, the anomaly detectors and dashboards — is a downstream decision that can be changed without touching the engine itself. That decoupling is the real product, not any individual element of the pipeline.
Three specific actions follow from this discussion: set acks=all and enable.idempotence=true on every producer; partition by hostname rather than timestamp; and always register schemas with a Schema Registry configured for BACKWARD compatibility. These three choices alone prevent the majority of outages observed on observability pipelines over many years. The remainder of this post represents optimisation and refinement — beneficial but not essential.
A final observation: this engine is a starting point rather than an endpoint. Once multivariate metrics flow reliably through Kafka, the substantive work begins — anomaly detection, capacity forecasting, automated remediation, and correlation with business metrics. Kafka is the unobtrusive, reliable infrastructure that enables all of this. When built carefully and left alone, it can operate quietly for years while more sophisticated systems are built on top of it.
References
- Apache Kafka Documentation—the canonical reference for broker configuration, topic design, and client semantics.
- confluent-kafka-python,the Python client used throughout this guide, built on librdkafka.
- Confluent Schema Registry—Avro schema management, compatibility modes, and evolution rules.
- Apache Avro Specification—the full schema language, including default values and schema resolution rules.
- psutil documentation,cross-platform process and system metric collection for Python.
- Kafka Producer Configuration Reference—authoritative list of every producer setting mentioned in this post.