[jira] [Commented] (KAFKA-4950) ConcurrentModificationException when iterating over Kafka Metrics

[Jean-Baptiste (JIRA)]

    [ 
https://issues.apache.org/jira/browse/KAFKA-4950?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16148437#comment-16148437
 ] 

Jean-Baptiste commented on KAFKA-4950:
--------------------------------------

Hi,

I was able to reproduce the issue using the following piece of code:
{code}
        // final KafkaConsumer<K, V> consumer = new KafkaConsumer<K, V>(props);
        // consumer.subscribe(....);

        new Thread(new Runnable() {

            @Override
            public void run() {
                while (true) {
                    for (Metric metric : consumer.metrics().values()) {
                        metric.value();
                    }
                }
            }
        }).start();

        while (true) {
            consumer.poll(100);
        }
{code}

I know that the consumer should not be used from several threads. However in my 
case I'm accessing them through the consumer JMX MBeans and then I have little 
control on when or from where they are called.

Here is an extract of the stacktrace I got when I access them from JMX MBeans:
{code}
javax.management.RuntimeMBeanException: 
java.util.ConcurrentModificationException
        at 
com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.rethrow(DefaultMBeanServerInterceptor.java:839)
        at 
com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.rethrowMaybeMBeanException(DefaultMBeanServerInterceptor.java:852)
        at 
com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.getAttribute(DefaultMBeanServerInterceptor.java:651)
        at 
com.sun.jmx.mbeanserver.JmxMBeanServer.getAttribute(JmxMBeanServer.java:678)
        ....
Caused by: java.util.ConcurrentModificationException
        at 
java.util.LinkedHashMap$LinkedHashIterator.nextNode(LinkedHashMap.java:719)
        at 
java.util.LinkedHashMap$LinkedKeyIterator.next(LinkedHashMap.java:742)
        at java.util.AbstractCollection.addAll(AbstractCollection.java:343)
        at java.util.HashSet.<init>(HashSet.java:119)
        at 
org.apache.kafka.common.internals.PartitionStates.partitionSet(PartitionStates.java:66)
        at 
org.apache.kafka.clients.consumer.internals.SubscriptionState.assignedPartitions(SubscriptionState.java:293)
        at 
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$ConsumerCoordinatorMetrics$1.measure(ConsumerCoordinator.java:884)
        at 
org.apache.kafka.common.metrics.KafkaMetric.value(KafkaMetric.java:61)
        at 
org.apache.kafka.common.metrics.KafkaMetric.value(KafkaMetric.java:52)
        at 
org.apache.kafka.common.metrics.JmxReporter$KafkaMbean.getAttribute(JmxReporter.java:183)
        at 
com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.getAttribute(DefaultMBeanServerInterceptor.java:647)
        ... 40 more
{code}


> ConcurrentModificationException when iterating over Kafka Metrics
> -----------------------------------------------------------------
>
>                 Key: KAFKA-4950
>                 URL: https://issues.apache.org/jira/browse/KAFKA-4950
>             Project: Kafka
>          Issue Type: Bug
>    Affects Versions: 0.10.1.1
>            Reporter: Dumitru Postoronca
>            Assignee: Vahid Hashemian
>            Priority: Minor
>             Fix For: 0.11.0.2
>
>
> It looks like the when calling {{PartitionStates.partitionSet()}}, while the 
> resulting Hashmap is being built, the internal state of the allocations can 
> change, which leads to ConcurrentModificationException during the copy 
> operation.
> {code}
> java.util.ConcurrentModificationException
>         at 
> java.util.LinkedHashMap$LinkedHashIterator.nextNode(LinkedHashMap.java:719)
>         at 
> java.util.LinkedHashMap$LinkedKeyIterator.next(LinkedHashMap.java:742)
>         at java.util.AbstractCollection.addAll(AbstractCollection.java:343)
>         at java.util.HashSet.<init>(HashSet.java:119)
>         at 
> org.apache.kafka.common.internals.PartitionStates.partitionSet(PartitionStates.java:66)
>         at 
> org.apache.kafka.clients.consumer.internals.SubscriptionState.assignedPartitions(SubscriptionState.java:291)
>         at 
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$ConsumerCoordinatorMetrics$1.measure(ConsumerCoordinator.java:783)
>         at 
> org.apache.kafka.common.metrics.KafkaMetric.value(KafkaMetric.java:61)
>         at 
> org.apache.kafka.common.metrics.KafkaMetric.value(KafkaMetric.java:52)
> {code}
> {code}
> // client code:
> import java.util.Collections;
> import java.util.HashMap;
> import java.util.Map;
> import com.codahale.metrics.Gauge;
> import com.codahale.metrics.Metric;
> import com.codahale.metrics.MetricSet;
> import org.apache.kafka.clients.consumer.KafkaConsumer;
> import org.apache.kafka.common.MetricName;
> import static com.codahale.metrics.MetricRegistry.name;
> public class KafkaMetricSet implements MetricSet {
>     private final KafkaConsumer client;
>     public KafkaMetricSet(KafkaConsumer client) {
>         this.client = client;
>     }
>     @Override
>     public Map<String, Metric> getMetrics() {
>         final Map<String, Metric> gauges = new HashMap<String, Metric>();
>         Map<MetricName, org.apache.kafka.common.Metric> m = client.metrics();
>         for (Map.Entry<MetricName, org.apache.kafka.common.Metric> e : 
> m.entrySet()) {
>             gauges.put(name(e.getKey().group(), e.getKey().name(), "count"), 
> new Gauge<Double>() {
>                 @Override
>                 public Double getValue() {
>                     return e.getValue().value(); // exception thrown here 
>                 }
>             });
>         }
>         return Collections.unmodifiableMap(gauges);
>     }
> }
> {code}



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)