[KAFKA-3632] ConsumerLag metrics persist after partition migration

• Type: Bug
• Status: Resolved
• Priority: Minor
• Resolution: Fixed
• Affects Version/s: 0.8.2.2, 0.9.0.1
• Fix Version/s: 0.9.0.2, 0.10.0.0
• Component/s: core
• Environment: JDK 1.8, Linux
• Assignee: Jason Gustafson
• Reporter: Brian Lueck
• Reviewer: Ismael Juma
• Created: 27/Apr/16 22:34
• Updated: 20/Jun/16 00:19
• Resolved: 04/May/16 19:11

Description

When a partition is migrated away from a broker, the ConsumerLag metric for the topic/partition gets 'stuck' at the current value. The only way to remove the metric is to restart the broker.

This appears to be because in AbstractFetcherThread.scala there is no way of removing a metric. See...

class FetcherLagStats(metricId: ClientIdAndBroker) { 
private val valueFactory = (k: ClientIdTopicPartition) => new FetcherLagMetrics(k) 
val stats = new Pool[ClientIdTopicPartition, FetcherLagMetrics](Some(valueFactory))

def getFetcherLagStats(topic: String, partitionId: Int): FetcherLagMetrics = { 
stats.getAndMaybePut(new ClientIdTopicPartition(metricId.clientId, topic, partitionId)) 
} 
}

Issue Links

• links to: GitHub Pull Request #1518