Kafka Client Throttling
Kafka implements throttling mechanisms to protect brokers from overload and ensure fair resource sharing among clients. This guide covers quota types, configuration, client-side backpressure, and quota monitoring.
Throttling Overview
Section titled “Throttling Overview”Broker-Side Quotas
Section titled “Broker-Side Quotas”Quota Types
Section titled “Quota Types”| Quota Type | Description | Unit |
|---|---|---|
producer_byte_rate | Upper bound for producer throughput | bytes/sec |
consumer_byte_rate | Upper bound for fetch throughput | bytes/sec |
request_percentage | Upper bound of time spent processing requests | percentage |
controller_mutation_rate | Rate limit for topic mutations | mutations/sec |
connection_creation_rate | Upper bound of connections accepted per IP | connections/sec |
Quota Hierarchy
Section titled “Quota Hierarchy”Configuration Examples
Section titled “Configuration Examples”Per-User Quota:
# Set quota for specific userkafka-configs.sh --bootstrap-server localhost:9092 \ --alter --add-config 'producer_byte_rate=10485760,consumer_byte_rate=20971520' \ --entity-type users --entity-name alicePer-Client-ID Quota:
# Set quota for specific client IDkafka-configs.sh --bootstrap-server localhost:9092 \ --alter --add-config 'producer_byte_rate=5242880' \ --entity-type clients --entity-name batch-producerUser + Client-ID Quota:
# Set quota for specific user and client combinationkafka-configs.sh --bootstrap-server localhost:9092 \ --alter --add-config 'producer_byte_rate=52428800' \ --entity-type users --entity-name alice \ --entity-type clients --entity-name high-priority-producerDefault Quota:
# Set default quota for all userskafka-configs.sh --bootstrap-server localhost:9092 \ --alter --add-config 'producer_byte_rate=1048576,consumer_byte_rate=2097152' \ --entity-type users --entity-default
# Set default quota for all clientskafka-configs.sh --bootstrap-server localhost:9092 \ --alter --add-config 'request_percentage=25' \ --entity-type clients --entity-defaultQuota Enforcement
Section titled “Quota Enforcement”How Throttling Works
Section titled “How Throttling Works”Throttle Time Calculation
Section titled “Throttle Time Calculation”throttle_time = (observed_rate - quota_limit) / quota_limit × window_sizeExample:
- Quota: 10 MB/s
- Observed rate: 15 MB/s
- Excess: 5 MB/s (50% over)
- Throttle time: ~500ms to bring back to limit
Window size comes from the quota sampling configuration (quota.window.size.seconds and num.quota.samples).
Client Response to Throttling
Section titled “Client Response to Throttling”Request Rate Quotas
Section titled “Request Rate Quotas”Request Handler Time
Section titled “Request Handler Time”# Percentage of I/O thread timerequest_percentage=25Calculation:
request_percentage is a percentage (double) representing the upper bound of time spent processing requests. Values can exceed 100 (for example, 200).Mutation Rate Quota (Kafka 2.7+)
Section titled “Mutation Rate Quota (Kafka 2.7+)”Limits admin operations to protect the controller:
# Set controller mutation rate quotakafka-configs.sh --bootstrap-server localhost:9092 \ --alter --add-config 'controller_mutation_rate=10' \ --entity-type users --entity-name admin-user| Operation | Mutations |
|---|---|
| CreateTopics (1 topic, 10 partitions) | 10 |
| DeleteTopics (1 topic) | 1 |
| CreatePartitions (+5 partitions) | 5 |
| AlterConfigs | Variable |
Client-Side Backpressure
Section titled “Client-Side Backpressure”Producer Backpressure
Section titled “Producer Backpressure”Configuration:
# Total buffer memorybuffer.memory=33554432 # 32 MB
# Max concurrent requests per connectionmax.in.flight.requests.per.connection=5
# Max time to block on send()max.block.ms=60000 # 60 secondsConsumer Backpressure
Section titled “Consumer Backpressure”Configuration:
# Max bytes per fetch responsefetch.max.bytes=52428800 # 50 MB
# Max bytes per partitionmax.partition.fetch.bytes=1048576 # 1 MB
# Max records per pollmax.poll.records=500Consumer Pause/Resume
Section titled “Consumer Pause/Resume”// Implement backpressure with pause/resumewhile (running) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (TopicPartition partition : records.partitions()) { if (shouldPause(partition)) { // Stop fetching from this partition consumer.pause(Collections.singleton(partition)); pausedPartitions.add(partition); } else { processRecords(records.records(partition)); } }
// Resume partitions that can accept more for (TopicPartition partition : pausedPartitions) { if (canResume(partition)) { consumer.resume(Collections.singleton(partition)); pausedPartitions.remove(partition); } }}Quota Monitoring
Section titled “Quota Monitoring”Broker Metrics
Section titled “Broker Metrics”| Metric | Description |
|---|---|
kafka.server:type=Fetch,name=queue-size | Throttled fetch queue size |
kafka.server:type=Produce,name=queue-size | Throttled produce queue size |
kafka.server:type=Request,name=queue-size | Throttled request queue size |
kafka.server:type=Fetch,name=throttle-time | Average fetch throttle time |
kafka.server:type=Produce,name=throttle-time | Average produce throttle time |
kafka.server:type=Request,name=throttle-time | Average request throttle time |
kafka.server:type=ControllerMutation,name=throttle-time | Average controller mutation throttle time |
Client Metrics
Section titled “Client Metrics”| Metric | Description |
|---|---|
produce-throttle-time-avg | Average produce throttle time |
produce-throttle-time-max | Max produce throttle time |
fetch-throttle-time-avg | Average fetch throttle time |
fetch-throttle-time-max | Max fetch throttle time |
Viewing Quotas
Section titled “Viewing Quotas”# List all quotaskafka-configs.sh --bootstrap-server localhost:9092 \ --describe --entity-type users
# Describe specific user quotakafka-configs.sh --bootstrap-server localhost:9092 \ --describe --entity-type users --entity-name alice
# Describe specific client quotakafka-configs.sh --bootstrap-server localhost:9092 \ --describe --entity-type clients --entity-name my-producerQuota Planning
Section titled “Quota Planning”Capacity Estimation
Section titled “Capacity Estimation”Quota Guidelines
Section titled “Quota Guidelines”| Client Type | Producer Quota | Consumer Quota | Request % |
|---|---|---|---|
| Critical real-time | 50-100 MB/s | 100-200 MB/s | 50% |
| Standard production | 10-50 MB/s | 50-100 MB/s | 25% |
| Batch processing | 5-20 MB/s | 20-50 MB/s | 10% |
| Development | 1-5 MB/s | 5-10 MB/s | 5% |
Dynamic Quota Updates
Section titled “Dynamic Quota Updates”Online Quota Changes
Section titled “Online Quota Changes”# Increase quota without restartkafka-configs.sh --bootstrap-server localhost:9092 \ --alter --add-config 'producer_byte_rate=52428800' \ --entity-type users --entity-name alice
# Remove quota (fall back to default)kafka-configs.sh --bootstrap-server localhost:9092 \ --alter --delete-config 'producer_byte_rate' \ --entity-type users --entity-name aliceQuota Change Propagation
Section titled “Quota Change Propagation”Handling Throttling
Section titled “Handling Throttling”Producer Handling
Section titled “Producer Handling”// Monitor throttle metricsproducer.metrics().forEach((name, metric) -> { if (name.name().contains("throttle")) { double value = (double) metric.metricValue(); if (value > 0) { log.warn("Producer throttled: {} = {}", name, value); } }});Consumer Handling
Section titled “Consumer Handling”// Check for throttling in consumerConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
// Get metricsMetric throttleMetric = consumer.metrics().get( new MetricName("fetch-throttle-time-avg", "consumer-fetch-manager-metrics", "", Collections.emptyMap()));
if (throttleMetric != null) { double throttleTime = (double) throttleMetric.metricValue(); if (throttleTime > 100) { log.warn("Consumer being throttled: {}ms average", throttleTime); }}Adaptive Rate Limiting
Section titled “Adaptive Rate Limiting”public class AdaptiveProducer { private final RateLimiter rateLimiter; private double currentRate;
public void send(ProducerRecord<String, String> record) { rateLimiter.acquire(); // Client-side rate limiting
producer.send(record, (metadata, exception) -> { if (metadata != null && metadata.throttleTimeMs() > 0) { // Reduce rate when throttled adjustRate(metadata.throttleTimeMs()); } }); }
private void adjustRate(long throttleTimeMs) { if (throttleTimeMs > 100) { currentRate *= 0.8; // Reduce by 20% rateLimiter.setRate(currentRate); } }}Network Bandwidth Throttling
Section titled “Network Bandwidth Throttling”Inter-Broker Throttling
Section titled “Inter-Broker Throttling”Throttle replication traffic during reassignment:
# Set replication throttlekafka-configs.sh --bootstrap-server localhost:9092 \ --alter --add-config 'leader.replication.throttled.rate=50000000,follower.replication.throttled.rate=50000000' \ --entity-type brokers --entity-name 1Throttle During Reassignment
Section titled “Throttle During Reassignment”# Reassignment with throttlekafka-reassign-partitions.sh --bootstrap-server localhost:9092 \ --reassignment-json-file reassignment.json \ --throttle 50000000 \ --execute
# Verify and remove throttle after completionkafka-reassign-partitions.sh --bootstrap-server localhost:9092 \ --reassignment-json-file reassignment.json \ --verifyTroubleshooting Throttling
Section titled “Troubleshooting Throttling”Diagnosing Throttling Issues
Section titled “Diagnosing Throttling Issues”| Symptom | Possible Cause | Solution |
|---|---|---|
| High throttle time | Quota too low | Increase quota |
| Inconsistent throttling | Multiple client IDs | Consolidate or set per-client quotas |
| Sudden throttling | Burst traffic | Implement client-side rate limiting |
| All clients throttled | Default quota too low | Increase default or set specific quotas |
Debug Commands
Section titled “Debug Commands”# Check current quotaskafka-configs.sh --bootstrap-server localhost:9092 \ --describe --entity-type users --entity-type clients
# Check broker metricskafka-run-class.sh kafka.tools.JmxTool \ --object-name 'kafka.server:type=ClientQuotaManager,user=*,client-id=*' \ --jmx-url service:jmx:rmi:///jndi/rmi://localhost:9999/jmxrmiVersion Compatibility
Section titled “Version Compatibility”| Feature | Minimum Version |
|---|---|
| Byte rate quotas | 0.9.0 |
| Request percentage quota | 0.11.0 |
| User quotas | 0.10.1 |
| User + Client-ID quotas | 0.10.1 |
| Controller mutation quota | 2.7.0 |
Best Practices
Section titled “Best Practices”Quota Configuration
Section titled “Quota Configuration”| Practice | Rationale |
|---|---|
| Set default quotas | Prevent unknown clients from overloading |
| Use user quotas with authentication | Accurate tracking per user |
| Leave headroom | Allow for traffic bursts |
| Monitor throttle metrics | Detect quota issues early |
Client Design
Section titled “Client Design”| Practice | Rationale |
|---|---|
| Implement client-side rate limiting | Smooth traffic patterns |
| Handle throttle response | Avoid wasted requests |
| Use pause/resume for consumers | Control processing rate |
| Monitor throttle metrics | Adapt to quota changes |
Related Documentation
Section titled “Related Documentation”- Kafka Protocol - Throttle time in responses
- Connection Pooling - Connection management
- Monitoring - Quota metrics
- Configuration - Full configuration reference