Skip to content

Reduce lock scope in Kafka metrics reporter #20535

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
laurit merged 5 commits into from
Oct 15, 2026
Merged

Reduce lock scope in Kafka metrics reporter #20535

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
db4cb4f
Initial plan
Copilot Oct 12, 2026
5ff0093
Reduce lock scope in Kafka metrics reporter
Copilot Oct 12, 2026
47f5e2b
Simplify Kafka observable registration removal
trask Oct 13, 2026
5b2df7c
Use ordered set for Kafka observable registrations
trask Oct 13, 2026
d16a281
Remove new Kafka metrics reporter tests
trask Oct 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 40 additions & 21 deletions ...etry/instrumentation/kafkaclients/common/v0_11/internal/OpenTelemetryMetricsReporter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
import io.opentelemetry.instrumentation.api.internal.EmbeddedInstrumentationProperties;
import io.opentelemetry.instrumentation.api.internal.GuardedBy;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
Expand Down Expand Up @@ -54,7 +54,7 @@ public final class OpenTelemetryMetricsReporter
private final Object lock = new Object();

@GuardedBy("lock")
private final List<RegisteredObservable> registeredObservables = new ArrayList<>();
private final Set<RegisteredObservable> registeredObservables = new LinkedHashSet<>();

/**
* Reset for test by resetting the {@link #meter} to {@code null} and closing all registered
Expand Down Expand Up @@ -98,45 +98,55 @@ public void metricChange(KafkaMetric metric) {
}

Set<AttributeKey<?>> attributeKeys = registeredObservable.getAttributes().asMap().keySet();
List<RegisteredObservable> observablesToClose = new ArrayList<>();
synchronized (lock) {
for (Iterator<RegisteredObservable> it = registeredObservables.iterator(); it.hasNext(); ) {
RegisteredObservable curRegisteredObservable = it.next();
for (RegisteredObservable curRegisteredObservable : registeredObservables) {
Set<AttributeKey<?>> curAttributeKeys =
curRegisteredObservable.getAttributes().asMap().keySet();
if (curRegisteredObservable.getKafkaMetricName().equals(metric.metricName())) {
logger.log(FINEST, "Replacing instrument: {0}", curRegisteredObservable);
closeInstrument(curRegisteredObservable.getObservable());
it.remove();
observablesToClose.add(curRegisteredObservable);
} else if (curRegisteredObservable
.getInstrumentDescriptor()
.equals(registeredObservable.getInstrumentDescriptor())
&& attributeKeys.size() > curAttributeKeys.size()
&& attributeKeys.containsAll(curAttributeKeys)) {
logger.log(
FINEST,
"Replacing instrument with higher dimension version: {0}",
curRegisteredObservable);
closeInstrument(curRegisteredObservable.getObservable());
it.remove();
observablesToClose.add(curRegisteredObservable);
}
}
}

for (RegisteredObservable curRegisteredObservable : observablesToClose) {
if (curRegisteredObservable.getKafkaMetricName().equals(metric.metricName())) {
logger.log(FINEST, "Replacing instrument: {0}", curRegisteredObservable);
} else {
logger.log(
FINEST,
"Replacing instrument with higher dimension version: {0}",
curRegisteredObservable);
}
closeInstrumentAndRemove(curRegisteredObservable);
}

synchronized (lock) {
registeredObservables.add(registeredObservable);
Comment thread
trask marked this conversation as resolved.
}
}

@Override
public void metricRemoval(KafkaMetric metric) {
logger.log(FINEST, "Metric removed: {0}", metric.metricName());
List<RegisteredObservable> observablesToClose = new ArrayList<>();
synchronized (lock) {
for (Iterator<RegisteredObservable> it = registeredObservables.iterator(); it.hasNext(); ) {
RegisteredObservable current = it.next();
for (RegisteredObservable current : registeredObservables) {
if (current.getKafkaMetricName().equals(metric.metricName())) {
closeInstrument(current.getObservable());
it.remove();
observablesToClose.add(current);
}
}
}

for (RegisteredObservable current : observablesToClose) {
closeInstrumentAndRemove(current);
}
}

@Override
Expand All @@ -163,11 +173,20 @@ public void onUpdate(ClusterResource clusterResource) {
}

private void closeAllInstruments() {
List<RegisteredObservable> observablesToClose;
synchronized (lock) {
for (Iterator<RegisteredObservable> it = registeredObservables.iterator(); it.hasNext(); ) {
closeInstrument(it.next().getObservable());
it.remove();
}
observablesToClose = new ArrayList<>(registeredObservables);
}

for (RegisteredObservable registeredObservable : observablesToClose) {
closeInstrumentAndRemove(registeredObservable);

@laurit laurit Oct 13, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Feels inefficient. It needs to loop over registeredObservables list to remove each element while previously it looped only once.

@trask trask Oct 13, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

5b2df7c

}
}

private void closeInstrumentAndRemove(RegisteredObservable registeredObservable) {
closeInstrument(registeredObservable.getObservable());
synchronized (lock) {
registeredObservables.remove(registeredObservable);
}
}

Expand Down
Loading