Skip to content

Commit 497b26d

Browse files
Fix PooledHashMap dropping live entries when an entry is removed during forEach (#8499)
1 parent 6d6bd70 commit 497b26d

3 files changed

Lines changed: 85 additions & 1 deletion

File tree

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/PooledHashMap.java

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -222,12 +222,14 @@ public void clear() {
222222
size = 0;
223223
}
224224

225+
// Walks each bucket back-to-front so the action can remove the entry it's currently on (as metric
226+
// collection does to drop stale series) without the next entry being skipped.
225227
@Override
226228
public void forEach(BiConsumer<? super K, ? super V> action) {
227229
for (int j = 0; j < table.length; j++) {
228230
ArrayList<Entry<K, V>> bucket = table[j];
229231
if (bucket != null) {
230-
for (int i = 0; i < bucket.size(); i++) {
232+
for (int i = bucket.size() - 1; i >= 0; i--) {
231233
Entry<K, V> entry = bucket.get(i);
232234
action.accept(entry.key, entry.value);
233235
}

sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/AsynchronousMetricStorageTest.java

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,9 @@
3939
import java.time.Duration;
4040
import java.time.Instant;
4141
import java.util.Collection;
42+
import java.util.LinkedHashSet;
43+
import java.util.Set;
44+
import java.util.stream.Collectors;
4245
import org.junit.jupiter.api.Test;
4346
import org.junit.jupiter.api.extension.ExtendWith;
4447
import org.junit.jupiter.api.extension.RegisterExtension;
@@ -413,6 +416,42 @@ void collect_CumulativeSeriesDisappearsAndReappears(MemoryMode memoryMode) {
413416
.hasValue(7)));
414417
}
415418

419+
@ParameterizedTest
420+
@EnumSource(MemoryMode.class)
421+
void collect_CumulativeRetainsLiveSeriesWhenStaleSeriesAreRemoved(MemoryMode memoryMode) {
422+
setup(memoryMode);
423+
424+
// Enough series to share buckets, but under the cardinality limit so there's no overflow
425+
// series.
426+
int seriesCount = 20;
427+
428+
// First collection reports everything.
429+
testClock.advance(Duration.ofSeconds(10));
430+
for (int i = 0; i < seriesCount; i++) {
431+
longCounterStorage.record(Attributes.builder().put("key", "v" + i).build(), 1);
432+
}
433+
longCounterStorage.collect(resource, scope, testClock.now());
434+
registeredReader.setLastCollectEpochNanos(testClock.now());
435+
436+
// Next time only the even series report; the odd ones go stale and get removed while iterating.
437+
// None of the live series should be skipped.
438+
testClock.advance(Duration.ofSeconds(10));
439+
Set<Attributes> expected = new LinkedHashSet<>();
440+
for (int i = 0; i < seriesCount; i += 2) {
441+
Attributes attributes = Attributes.builder().put("key", "v" + i).build();
442+
longCounterStorage.record(attributes, 2);
443+
expected.add(attributes);
444+
}
445+
446+
MetricData metricData = longCounterStorage.collect(resource, scope, testClock.now());
447+
Set<Attributes> collected =
448+
metricData.getLongSumData().getPoints().stream()
449+
.map(PointData::getAttributes)
450+
.collect(Collectors.toCollection(LinkedHashSet::new));
451+
452+
assertThat(collected).isEqualTo(expected);
453+
}
454+
416455
@ParameterizedTest
417456
@EnumSource(MemoryMode.class)
418457
void collect_DeltaComputesDiff(MemoryMode memoryMode) {

sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/PooledHashMapTest.java

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99

1010
import java.util.HashMap;
1111
import java.util.Map;
12+
import javax.annotation.Nullable;
1213
import org.junit.jupiter.api.BeforeEach;
1314
import org.junit.jupiter.api.Test;
1415

@@ -73,4 +74,46 @@ void forEachTest() {
7374

7475
assertThat(actualMap).containsOnlyKeys("One", "Two").containsValues(1, 2);
7576
}
77+
78+
@Test
79+
void forEachAllowsRemovalOfCurrentEntry() {
80+
// Keys with the same hashcode always share a bucket (independent of capacity), mirroring
81+
// collection removing a stale series while iterating the others.
82+
PooledHashMap<SameBucketKey, Integer> collidingMap = new PooledHashMap<>();
83+
SameBucketKey first = new SameBucketKey("first");
84+
SameBucketKey second = new SameBucketKey("second");
85+
collidingMap.put(first, 1);
86+
collidingMap.put(second, 2);
87+
88+
Map<SameBucketKey, Integer> visited = new HashMap<>();
89+
collidingMap.forEach(
90+
(key, value) -> {
91+
visited.put(key, value);
92+
if (visited.size() == 1) {
93+
collidingMap.remove(key); // drop the first one we see
94+
}
95+
});
96+
97+
assertThat(visited).containsOnlyKeys(first, second).containsValues(1, 2);
98+
assertThat(collidingMap.size()).isEqualTo(1);
99+
}
100+
101+
/** Key whose hashcode is constant, so all instances fall in the same bucket. */
102+
private static final class SameBucketKey {
103+
private final String name;
104+
105+
SameBucketKey(String name) {
106+
this.name = name;
107+
}
108+
109+
@Override
110+
public int hashCode() {
111+
return 1;
112+
}
113+
114+
@Override
115+
public boolean equals(@Nullable Object o) {
116+
return o instanceof SameBucketKey && name.equals(((SameBucketKey) o).name);
117+
}
118+
}
76119
}

0 commit comments

Comments
 (0)