Monitoring Broker Metrics

1. Understanding JMX Metrics

Metric Name	Description	Unit	Notes
kafka.server:type=KafkaServer,name=BrokerState	Current state of the broker (0 = Not Running, 1 = Starting, 2 = Running, 3 = Stopping, 4 = Stopped)	Integer	Used to monitor broker lifecycle
kafka.server:type=KafkaServer,name=ActiveControllerCount	Number of active controllers in the cluster	Integer	Should be 1 in a healthy cluster
kafka.server:type=KafkaServer,name=UnderReplicatedPartitions	Number of partitions that are under-replicated	Integer	Indicates replication issues
kafka.server:type=KafkaServer,name=OfflinePartitionsCount	Number of partitions that are offline	Integer	Indicates broker or topic issues
kafka.server:type=KafkaServer,name=UncleanLeaderElectionCount	Number of unclean leader elections	Integer	May indicate data loss risk
kafka.server:type=KafkaServer,name=BrokerTopicCount	Number of topics on this broker	Integer	Used for topic distribution monitoring

Example: Fetching JMX Metrics via JConsole

./jconsole <broker-host>:<port>

Note: JMX metrics are exposed via the kafka-server-start.sh script using -Dcom.sun.management.jmxremote JVM options.

Warning: Ensure JMX is secured in production environments to prevent unauthorized access.

2. Monitoring Bytes In/Out Rate

Metric	Description	Unit	Notes
bytes.in.rate	Rate of bytes received by the broker per second	bytes/sec	Measures incoming data rate to the broker
bytes.out.rate	Rate of bytes sent by the broker per second	bytes/sec	Measures outgoing data rate from the broker
bytes.in	Total bytes received by the broker since startup	bytes	Cumulative metric for incoming data
bytes.out	Total bytes sent by the broker since startup	bytes	Cumulative metric for outgoing data

Example: Kafka Broker Metrics Configuration

# Enable JMX Exporter for monitoring
export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote \
  -Dcom.sun.management.jmxremote.authenticate=false \
  -Dcom.sun.management.jmxremote.ssl=false \
  -Dcom.sun.management.jmxremote.port=9999"

# Example JMX metrics endpoint
# http://localhost:9999/jmx?query=Kafka.BrokerTopicMetrics:name=BytesInPerSec,topic=*

Note: These metrics are exposed via JMX and can be collected using tools like Prometheus or JConsole.

3. Tracking Message Rate

Metric	Description	Units	Notes
kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec	Number of messages received per second by the broker	messages/sec	Tracks all messages, including those that are filtered or rejected
kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec	Total bytes received per second by the broker	bytes/sec	Includes all bytes, including compression overhead
kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec	Total bytes sent per second by the broker	bytes/sec	Includes all bytes, including compression overhead

Example: Monitoring Message Rate via JMX

echo "get -b kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec" | jconsole

Note: The MessagesInPerSec metric is crucial for identifying message bursts or drops in throughput that may indicate broker overload or consumer lag.

4. Monitoring Request Metrics

Metric	Description	Unit	Notes
request.rate	Number of requests per second	requests/second	Measures the rate of incoming requests to the broker
request.total	Total number of requests received	requests	Cumulative count of all requests processed by the broker
request.size.avg	Average size of incoming requests	bytes	Helps identify potential large request issues
request.size.max	Maximum size of incoming requests	bytes	Used to detect unusually large requests that may cause performance issues
request.latency.avg	Average latency of request processing	milliseconds	Indicates the average time taken to process a request
request.latency.max	Maximum latency of request processing	milliseconds	Highlights the worst-case processing time for a request

Example: Configuring Request Metrics in Kafka

public class KafkaRequestMetrics {
    private final Metrics metrics;
    private final Sensor requestRateSensor;

    public KafkaRequestMetrics(Metrics metrics) {
        this.metrics = metrics;
        this.requestRateSensor = metrics.sensor("request-rate");
        this.requestRateSensor.add(new MetricName("request.rate", "kafka.server", "Number of requests per second"), new Rate());
    }

    public void recordRequest(long startTime, long endTime) {
        requestRateSensor.record(endTime - startTime);
    }
}

5. Checking Under-Replicated Partitions

Property	Description	Default Value	Notes
under_replicated_partitions	Number of partitions that are not fully replicated across all brokers	0	Indicates replication issues in the cluster

Example: Using Kafka Consumer Groups to Identify Under-Replicated Partitions

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group

Note: Under-replicated partitions can lead to data loss or reduced availability if not addressed promptly.

6. Monitoring ISR Shrink/Expand

Property	Description	Default Value	Notes
replica.lag.time.max.ms	Maximum time a replica may lag behind the leader before being considered out of sync.	`30000`	Used in ISR calculation. If a replica is lagging more than this time, it's removed from ISR.
replica.socket.receive.buffer.bytes	Size of the socket receive buffer for replicas.	`65536`	Affects performance and lag; can influence ISR stability.
replica.socket.timeout.ms	Socket timeout for replica fetch requests.	`30000`	Can cause ISR shrink if replicas are not responsive within this timeout.
replica.fetch.max.bytes	Maximum bytes a follower can fetch in one request.	`1048576`	Too small can cause lag, too large can cause timeouts.
replica.fetch.wait.max.ms	Maximum time a follower will wait for data before fetching.	`500`	Too long can cause ISR shrink due to perceived lag.

Example: ISR Shrink Alert Configuration

{
  "name": "ISR Shrink Alert",
  "description": "Triggers when ISR shrinks unexpectedly",
  "conditions": [
    {
      "metric": "kafka.server:type=ReplicaManager,name=ISRShrunk",
      "operator": "> 0",
      "threshold": 0
    }
  ],
  "actions": [
    {
      "type": "email",
      "to": "ops@example.com"
    }
  ]
}

Note: An ISR shrink indicates that replicas are falling behind the leader and are removed from the in-sync replica set. This can lead to data inconsistency if not addressed.

Warning: Frequent ISR shrink events may indicate hardware or network issues in the Kafka cluster.

7. Tracking Leader Election Rate

Metric	Description	Unit	Notes
kafka.controller.leader_election_rate	Number of leader elections per second	events/second	Indicates how frequently partitions are reassigned to different brokers

Example: Configuring Kafka to expose leader election rate

# Enable JMX for monitoring
export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote \
  -Dcom.sun.management.jmxremote.authenticate=false \
  -Dcom.sun.management.jmxremote.ssl=false"

# Start Kafka with JMX enabled
bin/kafka-server-start.sh config/server.properties

Note: High leader election rates can indicate broker instability or configuration issues.

8. Monitoring Log Flush Latency

Metric	Description	Unit	Default Aggregation
log.flush.ms	Time between flushes of log segments to disk	milliseconds	average
log.flush.offset.committed	Number of times the log flush offset was committed to ZooKeeper	count	sum
log.flush.scheduler.timeout.ms	Timeout for the log flush scheduler thread	milliseconds	average

Example: Configuring Log Flush Interval

{"log.flush.ms": 1000}

Note: Lower values improve durability but can impact performance.

9. Checking Network Processor Idle

Metric Name	Description	Unit	Notes
kafka.network:type=NetworkProcessor,name=IdlePercent	Percentage of time the network processor is idle	Percent	Indicates how much of the network processing time is unused

Example: Monitoring Idle Percent

kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic my-topic

Note: A consistently high idle percentage may indicate underutilization of network resources.

10. Monitoring Request Queue Size

Metric Name	Description	Unit	Notes
kafka.server:type=RequestQueueSize,name=RequestQueueSize	Number of requests currently queued for processing	requests	Indicates potential bottlenecks in request handling

Example: Monitoring Request Queue Size via JMX

echo -e 'get kafka.server:type=RequestQueueSize,name=RequestQueueSize' | jconsole

Note: A consistently high queue size may indicate that the broker is overwhelmed with requests and needs more resources or optimization.