Monitoring Consumer Metrics

1. Tracking Fetch Rate

Metric	Description	Unit	Notes
fetch.rate	Number of fetch requests per second	requests/second	Indicates how frequently the consumer is fetching data from brokers
fetch.latency.avg	Average time taken to fetch data	milliseconds	Measures the latency of fetch operations
fetch.throttle.time.avg	Average throttling time during fetches	milliseconds	Shows how much time is spent throttling due to broker limits

Example: Fetch Rate Monitoring

const kafka = new Kafka({ brokers: ['localhost:9092'] });
const consumer = kafka.consumer({ groupId: 'my-group' });

await consumer.connect();
await consumer.subscribe({ topic: 'my-topic' });

const metrics = await consumer.metrics();

console.log("Fetch Rate:", metrics.fetch.rate);
console.log("Avg Fetch Latency:", metrics.fetch.latency.avg);

2. Monitoring Records Consumed

Metric	Description	Unit	Notes
records.consumed.total	Total number of records consumed	count	Monotonically increasing counter
records.consumed.rate	Rate of records consumed per second	records/second	Calculated over a sliding window
records.lag	Number of records behind the latest offset	count	Indicates consumer lag

Example: Consuming Records with Metrics

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Properties;

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.METRICS_REPORTER_CLASSES_CONFIG, "org.apache.kafka.common.metrics.JmxReporter");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(java.util.Collections.singletonList("my-topic"));

while (true) {
    var records = consumer.poll(java.time.Duration.ofMillis(1000));
    for (var record : records) {
        System.out.printf("Consumed record: %s%n", record.value());
    }
}

3. Checking Consumer Lag

Example: Consumer Lag Monitoring with Kafka Consumer Groups

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

Field	Description	Notes
GROUP	Consumer group name	Identifier for the consumer group
CONSUMER-ID	Consumer instance ID	Unique identifier for each consumer in the group
CLIENT-ID	Client identifier	Application or client name
TOPIC	Topic name	Topic being consumed
PARTITION	Partition number	Partition ID within the topic
CURRENT-OFFSET	Current offset	Offset of the last message consumed
LOG-END-OFFSET	Log end offset	Latest offset in the partition
LAG	Consumer lag	Number of messages behind the latest offset
CONSUMER-ADDRESS	Consumer address	Network address of the consumer instance

Note: Consumer lag is calculated as LOG-END-OFFSET - CURRENT-OFFSET. High lag indicates potential performance issues or consumer overload.

4. Monitoring Fetch Latency

Metric	Description	Unit	Notes
fetch.latency.avg	Average time taken to fetch records from the broker	milliseconds	Indicates potential network or broker performance issues
fetch.latency.95th	95th percentile of fetch latency	milliseconds	Helps identify outliers in fetch performance
fetch.latency.99th	99th percentile of fetch latency	milliseconds	High percentile shows worst-case fetch delays

Example: Fetch Latency Configuration

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "consumer-group");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("session.timeout.ms", "30000");
props.put("fetch.min.bytes", "1");
props.put("fetch.max.wait.ms", "500"); // affects latency
props.put("fetch.max.bytes", "524288");
props.put("max.partition.fetch.bytes", "1048576");
props.put("client.id", "consumer-client");

Note: Fetch latency can be impacted by broker load, network conditions, and consumer configuration settings like fetch.max.wait.ms.

5. Tracking Commit Rate

Metric	Description	Unit	Notes
commit.rate	Rate of successful commits to the consumer group	commits/second	Indicates how frequently the consumer is committing offsets to the Kafka cluster
commit.latency.avg	Average time taken to commit offsets	milliseconds	Measures the latency of offset commits in the consumer group
commit.latency.max	Maximum time taken to commit offsets	milliseconds	Tracks the worst-case latency for offset commits in the consumer group

Example: Commit Rate Monitoring

const consumer = new KafkaConsumer({
  bootstrapServers: 'localhost:9092',
  groupId: 'my-group'
});

consumer.on('commit', (offsets) => {
  console.log(`Committed offsets: ${JSON.stringify(offsets)}`);
});

// Monitor commit rate via metrics
const commitRate = consumer.metrics().commit.rate;
console.log(`Commit Rate: ${commitRate} commits/sec`);

6. Monitoring Join Rate

Metric	Description	Unit	Notes
kafka.consumer.join.rate	Rate of consumer group joins per second	joins/second	Tracks how often consumers join a consumer group
kafka.consumer.join.total	Total number of consumer group joins	joins	Cumulative count of all joins since consumer started

Example: Consumer Group Join Rate Monitoring

const consumer = new KafkaConsumer({
  bootstrapServers: 'localhost:9092',
  groupId: 'my-group'
});

// Monitor join rate via JMX or Prometheus
const joinRate = consumer.metrics().joinRate;
console.log(`Join rate: ${joinRate} joins/sec`);

Note: High join rates may indicate consumer rebalances or group coordination issues.

7. Checking Sync Rate

Metric	Description	Unit	Notes
kafka.consumer.sync.rate	Rate of consumer sync operations per second	ops/sec	Indicates how frequently the consumer is synchronizing with the broker

Example: Monitoring Sync Rate

const consumer = new KafkaConsumer({
  bootstrapServers: 'localhost:9092',
  groupId: 'my-group'
});

consumer.on('sync', (syncInfo) => {
  console.log(`Sync rate: ${syncInfo.rate} ops/sec`);
});

8. Tracking Heartbeat Response Time

Metric Name	Description	Unit	Default Aggregation
consumer.heartbeat.response.time.ms	Time taken to receive a heartbeat response from the broker	Milliseconds	Average

Example: Configuring Heartbeat Response Time Monitoring

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("heartbeat.interval.ms", "3000");
props.put("session.timeout.ms", "45000");
props.put("metric.reporters", "org.apache.kafka.common.metrics.JmxReporter");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

Note: The heartbeat response time is critical for consumer group stability and can indicate broker performance issues or network latency.

9. Monitoring Last Poll Time

Metric Name	Description	Unit	Notes
last.poll.time.ms	Time in milliseconds since the last poll operation	ms	Indicates how long ago the consumer last polled for new messages

Example: Consumer Configuration

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.METRICS_REPORTERS_CONFIG, "io.confluent.metrics.reporter.ConfluentMetricsReporter");
props.put(ConsumerConfig.METRICS_SAMPLE_WINDOW_MS_CONFIG, 30000);
props.put(ConsumerConfig.METRICS_NUM_SAMPLES_CONFIG, 2);

Note: The last.poll.time.ms metric is useful for detecting consumer lag or potential issues in message processing.

10. Checking Assigned Partitions

Example: Consumer Group Assigned Partitions

const { Kafka } = require('kafkajs');

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['localhost:9092']
});

const consumer = kafka.consumer({ groupId: 'my-group' });

await consumer.connect();
await consumer.subscribe({ topic: 'my-topic', partitions: [0, 1, 2] });

// Get assigned partitions
const assignedPartitions = consumer.assignments();
console.log(assignedPartitions);

Property	Description	Values
topic	Name of the topic	`'my-topic'`
partition	Partition ID	`0`, `1`, `2`
offset	Current offset position	`'0'`, `'100'`

Note: Use consumer.assignments() to check which partitions are currently assigned to a consumer in a group.