Monitoring Producer Metrics

1. Tracking Record Send Rate

Property	Description	Default Value	Notes
record.send.rate	Number of records sent per second	0	Measures throughput of record production
records.sent.total	Total number of records sent	0	Cumulative count since producer started

Example: Monitoring send rate with JMX

echo "get record.send.rate" | jconsole

Note: The record.send.rate metric is useful for identifying producer bottlenecks or bursts in data flow.

2. Monitoring Batch Size

Property	Default Value	Description	Notes
batch.size	`16384`	Maximum size of a batch in bytes	Controls how many messages are grouped together before sending to Kafka
max.request.size	`1048576`	Maximum size of a request in bytes	Must be larger than or equal to `batch.size`

Example: Configuring Batch Size

{
  "bootstrap.servers": "localhost:9092",
  "batch.size": 32768,
  "linger.ms": 5
}

Note: Increasing batch.size can improve throughput but may increase latency due to waiting for larger batches.

3. Checking Compression Ratio

Property	Value	Description
compression.type	`none`, `gzip`, `snappy`, `lz4`, `zstd`	Compression type to use for messages
compression.level	`0` to `9` (gzip), `1` to `4` (snappy/lz4)	Compression level (higher = better ratio, more CPU)

Example: Configuring Compression

{
  "compression.type": "gzip",
  "compression.level": 6
}

Note: Compression ratio depends on message content and type. Gzip typically provides the best compression ratio for text-based messages.

4. Monitoring Record Queue Time

Metric	Description	Unit	Notes
record.queue.time.ms	Average time a record spends in the producer's internal queue before being sent to the broker	Milliseconds	Indicates potential bottlenecks in producer throughput or broker responsiveness
record.queue.time.ms.max	Maximum queue time for any record in the current sampling window	Milliseconds	Helps identify outliers or bursts in queueing delays

Example: Configuring Producer Metrics

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.METRIC_REPORTERS_CONFIG, "io.confluent.metrics.reporter.ConfluentMetricsReporter");
props.put(ProducerConfig.METRICS_SAMPLE_WINDOW_MS_CONFIG, 30000);
props.put(ProducerConfig.METRICS_NUM_SAMPLES_CONFIG, 2);

Note: The record.queue.time.ms metric is particularly useful for identifying producer bottlenecks when records are queued but not yet sent due to broker unavailability or high load.

5. Tracking Request Latency

Metric	Description	Unit	Notes
request.latency.avg	Average time taken to send a request to the broker	milliseconds	Measures end-to-end latency including network overhead
request.latency.max	Maximum time taken to send a request to the broker	milliseconds	Identifies outliers in request processing
request.size.avg	Average size of requests sent to the broker	bytes	Helps in understanding payload distribution

Example: Monitoring Request Latency with JMX

bin/kafka-run-class.sh kafka.tools.JmxTool \
  --reporting-interface csv \
  --jmx-url service:jmx:rmi:///jndi/rmi://localhost:9999/jmxrmi \
  --output-file latency_metrics.csv \
  --interval 5

Note: Latency metrics are crucial for identifying bottlenecks in producer performance. High latency can indicate network issues or broker overload.

6. Monitoring Buffer Available Bytes

Property	Description	Default Value	Notes
buffer.available.bytes	Number of bytes available in the producer's internal buffer	Unlimited (based on memory)	Indicates how much space is available for new records before the producer starts blocking or rejecting new messages

Example: Configuring Buffer Size

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("buffer.memory", 33554432); // 32 MB
props.put("batch.size", 16384); // 16 KB

Note: The buffer.memory setting controls the total size of the buffer pool available to the producer for buffering records before they are sent to Kafka brokers.

7. Checking Waiting Threads

Property	Description	Default Value	Notes
producer.waiting.threads.count	Number of threads currently waiting for a response from the Kafka broker.	0	Used to monitor producer thread pool saturation.
producer.waiting.threads.max	Maximum number of threads that have waited for a response from the Kafka broker.	0	Indicates peak thread waiting activity.

Example: Monitoring Waiting Threads

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import java.util.Properties;

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

// Example of checking thread waiting metrics via JMX or Prometheus
System.out.println("Current waiting threads: " + producer.metrics().get("waiting-threads-count").value());

Note: The waiting-threads-count metric is useful for identifying potential bottlenecks in producer throughput. If this value is consistently high, it may indicate that the producer is not keeping up with the broker's response rate.

8. Tracking Record Error Rate

Property	Description	Default Value	Notes
record.error.rate	Measures the rate of record processing errors per second	0.0	Tracks failed message sends or processing errors
record.error.total	Total count of record errors since producer start	0	Cumulative error counter for monitoring

Example: Error Rate Monitoring

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("metric.reporters", "com.example.KafkaMetricsReporter");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

// Example of error handling and logging
producer.send(new ProducerRecord<String, String>("test-topic", "key1", "value1"),
    (metadata, exception) -> {
        if (exception != null) {
            System.err.println("Error sending record: " + exception.getMessage());
        }
    });

Note: The record.error.rate metric is useful for identifying issues in message delivery or serialization failures.

9. Monitoring Record Retry Rate

Property	Description	Default Value	Notes
retries	Number of times to retry sending a record before giving up	`2147483647`	Set to `0` to disable retries
retry.backoff.ms	Time in milliseconds to wait between retries	`100`	Exponential backoff is applied
retry.backoff.max.ms	Maximum time in milliseconds to wait between retries	`1000`	Used to cap exponential backoff

Example: Configuring Retry Rate

{
  "retries": 3,
  "retry.backoff.ms": 200,
  "retry.backoff.max.ms": 1000
}

Note: Retry rate can be monitored via record.retry.rate and record.error.rate metrics in Kafka Producer.

10. Checking Metadata Age

Property	Value	Description
kafka.producer.metadata.age.ms	`30000`	Time in milliseconds after which metadata is considered stale and refreshed.
kafka.producer.metadata.max.age.ms	`300000`	Maximum time in milliseconds before metadata is forcefully refreshed.

Example: Configuring Metadata Age

{
  "bootstrap.servers": "localhost:9092",
  "metadata.max.age.ms": 300000,
  "metadata.age.ms": 30000
}