ConcurrentModifiedException when read metrics from Kafka clients.

[Cirilla-zmh]

Describe the bug

TraceStack:

msg:"An exception occurred invoking callback for CallbackRegistration{instrumentDescriptors=[InstrumentDescriptor{name=kafka.consumer.assigned_partitions, description=The number of partitions currently assigned to this consumer, unit=, type=OBSERVABLE_GAUGE, valueType=DOUBLE, advice=Advice{explicitBucketBoundaries=null}}]}. java.util.ConcurrentModificationException
	at java.base/java.util.LinkedHashMap$LinkedHashIterator.nextNode(LinkedHashMap.java:719)
	at java.base/java.util.LinkedHashMap$LinkedKeyIterator.next(LinkedHashMap.java:741)
	at java.base/java.util.AbstractCollection.addAll(AbstractCollection.java:351)
	at java.base/java.util.HashSet.<init>(HashSet.java:120)
	at org.apache.kafka.common.internals.PartitionStates.partitionSet(PartitionStates.java:65)
	at org.apache.kafka.clients.consumer.internals.SubscriptionState.assignedPartitions(SubscriptionState.java:298)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$ConsumerCoordinatorMetrics$1.measure(ConsumerCoordinator.java:896)
	at io.opentelemetry.javaagent.shaded.instrumentation.kafka.internal.KafkaMetricRegistry.value(KafkaMetricRegistry.java:145)
	at io.opentelemetry.javaagent.shaded.instrumentation.kafka.internal.KafkaMetricRegistry.lambda$createObservable$2(KafkaMetricRegistry.java:124)
	at io.opentelemetry.sdk.metrics.AbstractInstrumentBuilder.lambda$registerDoubleAsynchronousInstrument$0(AbstractInstrumentBuilder.java:107)
	at io.opentelemetry.sdk.metrics.internal.state.CallbackRegistration.invokeCallback(CallbackRegistration.java:84)
	at io.opentelemetry.sdk.metrics.internal.state.MeterSharedState.collectAll(MeterSharedState.java:96)
	at io.opentelemetry.sdk.metrics.SdkMeter.collectAll(SdkMeter.java:75)
	at io.opentelemetry.sdk.metrics.SdkMeterProvider$LeasedMetricProducer.collectAllMetrics(SdkMeterProvider.java:184)
	......
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
	at java.base/java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305)
	at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
	at java.base/java.lang.Thread.run(Thread.java:829)

Steps to reproduce

In fact, this is not a bug that belongs to the instrumentation project; it was only triggered by the instrumentation.

Expected behavior

A ConcurrentModifiedException will be thrown at the metric collection point, which will cause a single metric collection to fail.

Actual behavior

The OpenTelemetryMetricsReporter instances should not be injected into Kafka clients with a version lower than 2.0.1. (See the link; version 2.0.1 of kafka-clients has fixed this issue.)

Javaagent or library instrumentation version

v2.5.0

Environment

JDK: HotSpot 17.0.7
OS: Darwin 22.3.0

Additional context

No response

[Cirilla-zmh]

I believe that users of Java Agent with Kafka versions lower than 2.0.1 may encounter the same issue. It would be wise to skip the injection of OpenTelemetryMetricsReporter if a Kafka version lower than 2.0.1 is detected. At the least, a specific note should highlight this issue.

As a redeveloper of opentelemetry-java-instrumentation, I am interested in knowing if there are any effective methods to predict these issues before they cause problems.

[Cirilla-zmh]

Here is the complementary details:

While fetching metric named "assigned-partitions", subscriptions.assignedPartitions().size() will be called.

private class ConsumerCoordinatorMetrics {
    private final String metricGrpName;
    private final Sensor commitLatency;

    private ConsumerCoordinatorMetrics(Metrics metrics, String metricGrpPrefix) {
        this.metricGrpName = metricGrpPrefix + "-coordinator-metrics";

        this.commitLatency = metrics.sensor("commit-latency");
        this.commitLatency.add(metrics.metricName("commit-latency-avg",
            this.metricGrpName,
            "The average time taken for a commit request"), new Avg());
        this.commitLatency.add(metrics.metricName("commit-latency-max",
            this.metricGrpName,
            "The max time taken for a commit request"), new Max());
        this.commitLatency.add(createMeter(metrics, metricGrpName, "commit", "commit calls"));

        Measurable numParts =
            new Measurable() {
                public double measure(MetricConfig config, long now) {
                    return subscriptions.assignedPartitions().size();
                }
            };
        metrics.addMetric(metrics.metricName("assigned-partitions",
            this.metricGrpName,
            "The number of partitions currently assigned to this consumer"), numParts);
    }
}

However, the assignedPartitions() method would return a deep copy of partitionSet, created by iterating the key set of partitionMap. While the partitionMap is modified, the action of fetching metrics will throw an ConcurrentModifiedException.

public Set<TopicPartition> partitionSet() {
    return new HashSet<>(map.keySet());
}

[trask]

The OpenTelemetryMetricsReporter instances should not be injected into Kafka clients with a version lower than 2.0.1. (See the link; version 2.0.1 of kafka-clients has fixed this issue.)

hi @Cirilla-zmh! can you reproduce this in OpenTelemetryMetricsReporterTest.java?

I looked at the upstream fix a bit more.

I think it would be ok to restrict OpenTelemetryMetricsReporter instrumentation to kafka-clients 2.0.1+.

@laurit or @jack-berg do I remember you suggesting to disable kafka-clients metrics instrumentation by default (similar to what we did for micrometer metrics) since it's just a bridge as opposed to semconv metrics? I think that could be a good idea for 3.0, and we could restrict OpenTelemetryMetricsReporter instrumentation to kafka-clients 2.0.1+ at that time.

[laurit]

Another issue that users have reported with kafka metrics is that when they have multiple producers/consumers which have the same client id they end up with duplicate metrics. This is because each producer/consumer produces the same metrics with the same attributes.