From e45bd731daadc093c8f2294618633c6edefb3b8b Mon Sep 17 00:00:00 2001 From: Gregor Zeitlinger Date: Tue, 15 Sep 2026 11:12:17 +0000 Subject: [PATCH 1/2] test: focus PR benchmarks on client_java and measure label lookups Signed-off-by: Gregor Zeitlinger --- .github/workflows/lint.yml | 6 ++ .github/workflows/pr-benchmarks.yml | 4 +- .mise/tasks/generate_benchmark_summary.py | 64 ++++++++++++++- .../tasks/test_generate-benchmark-summary.py | 59 ++++++++++++++ .mise/tasks/test_pr-benchmark-selection.py | 79 +++++++++++++++++++ benchmarks/README.md | 29 +++++++ .../metrics/benchmarks/CounterBenchmark.java | 42 ++++++++++ 7 files changed, 277 insertions(+), 6 deletions(-) create mode 100644 .mise/tasks/test_pr-benchmark-selection.py diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index bfb5b15172..70762dde72 100644 --- a/.github/workflows/lint.yml +++ b/.github/workflows/lint.yml @@ -34,3 +34,9 @@ jobs: GITHUB_HEAD_REF: ${{ github.head_ref }} PR_HEAD_REPO: ${{ github.event.pull_request.head.repo.full_name }} run: mise run lint + + - name: Test benchmark tooling + run: | + for test_file in .mise/tasks/test_*.py; do + python3 -B "$test_file" + done diff --git a/.github/workflows/pr-benchmarks.yml b/.github/workflows/pr-benchmarks.yml index f136cd4f37..d96424f76f 100644 --- a/.github/workflows/pr-benchmarks.yml +++ b/.github/workflows/pr-benchmarks.yml @@ -24,9 +24,9 @@ jobs: matrix: include: - topic: counter - pattern: CounterBenchmark + pattern: 'CounterBenchmark[.]prometheus.*' - topic: histogram - pattern: HistogramBenchmark + pattern: 'HistogramBenchmark[.]prometheus.*' - topic: exposition pattern: HistogramTextFormatBenchmark|TextFormatUtilBenchmark permissions: diff --git a/.mise/tasks/generate_benchmark_summary.py b/.mise/tasks/generate_benchmark_summary.py index 0da622bcbf..a7fc32d07d 100755 --- a/.mise/tasks/generate_benchmark_summary.py +++ b/.mise/tasks/generate_benchmark_summary.py @@ -470,6 +470,60 @@ def generate_comparison_section( return md +def allocation_score(result: dict) -> float | None: + """Read normalized GC allocation, accepting zero but not missing/invalid data.""" + metric = result.get("secondaryMetrics", {}).get("gc.alloc.rate.norm", {}) + if metric.get("scoreUnit") != "B/op": + return None + try: + score = float(metric.get("score")) + except (TypeError, ValueError): + return None + return score if math.isfinite(score) and score >= 0 else None + + +def generate_allocation_section(results: list, baseline_results: list) -> list[str]: + """Show allocation separately from the throughput/latency regression verdict.""" + baseline_by_name = {b.get("benchmark", ""): b for b in baseline_results} + rows = [] + for head in sorted(results, key=lambda b: b.get("benchmark", "")): + name = head.get("benchmark", "") + baseline = baseline_by_name.get(name, {}) + head_score = allocation_score(head) + base_score = allocation_score(baseline) + if head_score is None and base_score is None: + continue + head_text = "—" if head_score is None else f"{head_score:.3f}" + base_text = "—" if base_score is None else f"{base_score:.3f}" + change = "—" + if ( + head_score is not None + and base_score is not None + and comparable_metadata(head, baseline) + ): + change = f"{head_score - base_score:+.3f}" + rows.append( + f"| {short_benchmark_name(name)} | {head_text} | {base_text} | {change} |" + ) + if not rows: + return [] + return [ + "## Allocation per operation", + "", + "JMH GC profiler `gc.alloc.rate.norm`, in bytes per benchmark operation (lower is better).", + ( + "Delta is PR minus base, shown only for matching benchmark configurations. " + "Values are descriptive, not statistical regression verdicts; " + "— means unavailable or not comparable. Each benchmark defines its own operation." + ), + "", + "| Benchmark | PR B/op | Base B/op | Delta B/op |", + "|:----------|--------:|----------:|-----------:|", + *rows, + "", + ] + + def generate_markdown( results: list, commit_sha: str, @@ -488,7 +542,7 @@ def generate_markdown( first = results[0] if results else {} jdk_version = first.get("jdkVersion", "unknown") vm_name = first.get("vmName", "unknown") - threads = first.get("threads", "?") + threads = "/".join(sorted({str(b.get("threads", "?")) for b in results})) or "?" forks = first.get("forks", "?") warmup_iters = first.get("warmupIterations", "?") measure_iters = first.get("measurementIterations", "?") @@ -619,9 +673,11 @@ def generate_markdown( ) md.append("") + md.extend(generate_allocation_section(results, baseline_results or [])) + md.append("### Raw Results") md.append("") - md.append("```") + md.append("```text") md.append( f"{'Benchmark':<50} {'Mode':>6} {'Cnt':>4} {'Score':>14} {'Error':>12} Units" ) @@ -680,8 +736,8 @@ def generate_markdown( md.append("| Benchmark | Description |") md.append("|:----------|:------------|") md.append( - "| **CounterBenchmark** | Counter increment performance: " - "Prometheus, OpenTelemetry, simpleclient, Codahale |" + "| **CounterBenchmark** | Counter updates and label-value lookup " + "(selected methods only) |" ) md.append( "| **HistogramBenchmark** | Histogram observation performance " diff --git a/.mise/tasks/test_generate-benchmark-summary.py b/.mise/tasks/test_generate-benchmark-summary.py index 9918d0e49c..d46ee23ea2 100644 --- a/.mise/tasks/test_generate-benchmark-summary.py +++ b/.mise/tasks/test_generate-benchmark-summary.py @@ -7,7 +7,9 @@ sys.path.insert(0, here) from generate_benchmark_summary import ( + allocation_score, comparison_status, + generate_allocation_section, generate_markdown, ) @@ -45,6 +47,63 @@ def result( } +def with_allocation(benchmark, score, unit="B/op"): + benchmark["secondaryMetrics"] = { + "gc.alloc.rate.norm": {"score": score, "scoreUnit": unit} + } + return benchmark + + +class TestAllocationSummary(unittest.TestCase): + def test_zero_is_valid_but_missing_invalid_and_wrong_units_are_not(self): + self.assertEqual(allocation_score(with_allocation(result(), 0)), 0) + self.assertIsNone(allocation_score(result())) + self.assertIsNone(allocation_score(with_allocation(result(), 1, "MB/sec"))) + for score in (None, "NaN", float("inf"), -1, "not a number"): + self.assertIsNone(allocation_score(with_allocation(result(), score))) + + def test_allocation_delta_is_absolute_and_lower_is_better(self): + head = with_allocation(result(), 0) + base = with_allocation(result(), 16) + section = "\n".join(generate_allocation_section([head], [base])) + self.assertIn("| 0.000 | 16.000 | -16.000 |", section) + self.assertIn("not statistical regression verdicts", section) + + def test_missing_or_incomparable_base_has_no_delta(self): + head = with_allocation(result(), 16) + for base in ([], [result()], [with_allocation(result(threads=1), 32)]): + section = "\n".join(generate_allocation_section([head], base)) + self.assertTrue(section.rstrip().endswith("| — |")) + self.assertIn( + "| 16.000 | — | — |", + "\n".join(generate_allocation_section([head], [])), + ) + + def test_missing_head_allocation_is_not_reported_as_zero(self): + section = "\n".join( + generate_allocation_section([result()], [with_allocation(result(), 16)]) + ) + self.assertIn("| — | 16.000 | — |", section) + + def test_no_gc_data_omits_allocation_section(self): + self.assertEqual(generate_allocation_section([result()], []), []) + + def test_markdown_includes_head_only_allocations_and_mixed_threads(self): + base = result() + head = with_allocation(result(name="CounterBenchmark.newLookup", threads=1), 24) + markdown = generate_markdown( + [base, head], + "head", + "prometheus/client_java", + [base], + "base", + "prometheus/client_java", + ) + self.assertIn("## Allocation per operation", markdown) + self.assertIn("| CounterBenchmark.newLookup | 24.000 | — | — |", markdown) + self.assertIn("1/4 threads", markdown) + + class TestBenchmarkComparison(unittest.TestCase): def test_meaningful_improvement_requires_threshold_and_separation(self): self.assertEqual( diff --git a/.mise/tasks/test_pr-benchmark-selection.py b/.mise/tasks/test_pr-benchmark-selection.py new file mode 100644 index 0000000000..63befebf78 --- /dev/null +++ b/.mise/tasks/test_pr-benchmark-selection.py @@ -0,0 +1,79 @@ +import re +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +WORKFLOW = (ROOT / ".github/workflows/pr-benchmarks.yml").read_text() +PATTERNS = { + topic: pattern.strip("'\"") + for topic, pattern in re.findall(r"- topic: (\w+)\s+pattern: ([^\n]+)", WORKFLOW) +} +PACKAGE = "io.prometheus.metrics.benchmarks." + + +class TestPrBenchmarkSelection(unittest.TestCase): + def test_only_client_java_counter_and_histogram_methods_are_selected(self): + for topic, class_name in ( + ("counter", "CounterBenchmark"), + ("histogram", "HistogramBenchmark"), + ): + source = ( + ROOT + / "benchmarks/src/main/java/io/prometheus/metrics/benchmarks" + / f"{class_name}.java" + ).read_text() + methods = re.findall( + r"@Benchmark\s+@Threads\(\d+\)\s+public \S+ (\w+)\(", source + ) + self.assertTrue(methods) + for method in methods: + with self.subTest(method=method): + selected = re.search( + PATTERNS[topic], PACKAGE + class_name + "." + method + ) + self.assertEqual( + selected is not None, method.startswith("prometheus") + ) + + def test_external_systems_including_future_bound_instruments_are_excluded(self): + for class_name in ("CounterBenchmark", "HistogramBenchmark"): + for method in ( + "openTelemetryAdd", + "openTelemetryBoundInc", + "openTelemetryBoundClassic", + "codahaleIncNoLabels", + "simpleclientAdd", + "simpleclient", + ): + name = PACKAGE + class_name + "." + method + self.assertFalse( + any(re.search(p, name) for p in PATTERNS.values()), name + ) + + def test_all_lookup_and_cached_variants_are_selected(self): + for method in ( + "prometheusLabelValuesInc", + "prometheusLabelValuesIncSingleThread", + "prometheusCachedLabelValuesInc", + "prometheusCachedLabelValuesIncSingleThread", + ): + self.assertRegex( + PACKAGE + "CounterBenchmark." + method, PATTERNS["counter"] + ) + + def test_exposition_keeps_openmetrics_and_prometheus_formats(self): + for class_name in ("HistogramTextFormatBenchmark", "TextFormatUtilBenchmark"): + for method in ("openMetricsWriteToNull", "prometheusWriteToNull"): + self.assertRegex( + PACKAGE + class_name + "." + method, PATTERNS["exposition"] + ) + + def test_base_and_head_use_the_same_selection(self): + self.assertIn("JMH_PATTERN: ${{ matrix.pattern }}", WORKFLOW) + self.assertEqual( + WORKFLOW.count("JMH_ARGS: -f 3 -wi 3 -i 5 ${{ env.JMH_PATTERN }}"), 2 + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/benchmarks/README.md b/benchmarks/README.md index 3859eddea1..0d70149134 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -55,6 +55,35 @@ JMH parameter reference: ## Results +### Pull request benchmarks + +The `benchmark` label runs the PR head and base on the same runner for each topic. +PR runs select only client_java counter and histogram methods (`prometheus*`), plus +the exposition benchmarks. OpenTelemetry, Codahale, and legacy simpleclient methods +are excluded from PR runs, but remain available in the full/local and nightly suites. +OpenMetrics exposition remains included: it is a client_java output format. + +The `CounterBenchmark.prometheusLabelValuesInc*` methods repeatedly look up an +existing label combination and increment it. The matching +`prometheusCachedLabelValuesInc*` methods increment a cached data point instead. +Both have one-thread and four-thread variants sharing a counter. Each invocation +performs one metric update, so throughput and GC profiler allocation in B/op are +per update, unlike older benchmarks that batch updates in a loop. + +Run just the lookup and cached variants with allocation profiling: + +```shell +./mvnw -pl benchmarks -am package -DskipTests +java -jar benchmarks/target/benchmarks.jar \ + 'CounterBenchmark[.]prometheus(Cached)?LabelValuesInc.*' \ + -f 3 -wi 3 -i 5 -prof gc +``` + +The PR report shows allocation separately from throughput. Allocation deltas are +descriptive, not statistical verdicts, and require matching base/head configurations. +New benchmarks initially have head-only results. To evaluate a production change, +run the same benchmark source and JVM configuration against both implementations. + See Javadoc of the benchmark classes: - [CounterBenchmark](https://github.com/prometheus/client_java/blob/main/benchmarks/src/main/java/io/prometheus/metrics/benchmarks/CounterBenchmark.java) diff --git a/benchmarks/src/main/java/io/prometheus/metrics/benchmarks/CounterBenchmark.java b/benchmarks/src/main/java/io/prometheus/metrics/benchmarks/CounterBenchmark.java index d8b75e437c..06d2648f98 100644 --- a/benchmarks/src/main/java/io/prometheus/metrics/benchmarks/CounterBenchmark.java +++ b/benchmarks/src/main/java/io/prometheus/metrics/benchmarks/CounterBenchmark.java @@ -55,6 +55,48 @@ public PrometheusCounter() { } } + /** Pre-populated labels so lookup benchmarks measure hits, not data point creation. */ + @State(Scope.Benchmark) + public static class PrometheusLabelLookup { + final Counter counter = + Counter.builder().name("lookup_test").labelNames("path", "status").build(); + final String path = "/"; + final String status = "200"; + final CounterDataPoint cached = counter.labelValues(path, status); + } + + // Each invocation performs one increment, so GC profiler B/op is per metric update. Keep the + // same work and shared counter in the lookup and cached variants; only label resolution differs. + @Benchmark + @Threads(1) + public CounterDataPoint prometheusLabelValuesIncSingleThread(PrometheusLabelLookup state) { + CounterDataPoint dataPoint = state.counter.labelValues(state.path, state.status); + dataPoint.inc(); + return dataPoint; + } + + @Benchmark + @Threads(4) + public CounterDataPoint prometheusLabelValuesInc(PrometheusLabelLookup state) { + CounterDataPoint dataPoint = state.counter.labelValues(state.path, state.status); + dataPoint.inc(); + return dataPoint; + } + + @Benchmark + @Threads(1) + public CounterDataPoint prometheusCachedLabelValuesIncSingleThread(PrometheusLabelLookup state) { + state.cached.inc(); + return state.cached; + } + + @Benchmark + @Threads(4) + public CounterDataPoint prometheusCachedLabelValuesInc(PrometheusLabelLookup state) { + state.cached.inc(); + return state.cached; + } + @State(Scope.Benchmark) public static class SimpleclientCounter { From 5da5c49dffec74e4b25785f9c69ad24bf33a7e41 Mon Sep 17 00:00:00 2001 From: Gregor Zeitlinger Date: Tue, 15 Sep 2026 11:36:29 +0000 Subject: [PATCH 2/2] fix: keep late observations out of subsequent collection buffers Signed-off-by: Gregor Zeitlinger --- .../metrics/core/metrics/Buffer.java | 28 ++- .../metrics/core/metrics/BufferTest.java | 175 +++++++++++------- 2 files changed, 130 insertions(+), 73 deletions(-) diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java index 6dc68f8e6b..22525840c9 100644 --- a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java +++ b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java @@ -49,6 +49,10 @@ private static final class Generation { // available processors. This is simpler than the striping used by LongAdder, so hot spots remain // possible when several recording threads resolve to the same stripe. private final AtomicLong[] stripedObservationCounts; + // Protected by appendLock. These are absolute per-stripe observation counts at activation, not + // the reset-adjusted count used by complete. Reused across generations to avoid scrape + // allocations. + private final long[] generationStartCounts; private final ReentrantLock observationLock = new ReentrantLock(); private boolean reset; private long observationCountOffset; @@ -76,33 +80,41 @@ private static final class Generation { this.maxBufferSize = maxBufferSize; this.beforeAppendLock = beforeAppendLock; stripedObservationCounts = new AtomicLong[Runtime.getRuntime().availableProcessors()]; + generationStartCounts = new long[stripedObservationCounts.length]; for (int i = 0; i < stripedObservationCounts.length; i++) { stripedObservationCounts[i] = new AtomicLong(); } } boolean append(double value) { - AtomicLong counter = - stripedObservationCounts[ - stripeIndex(Thread.currentThread().getId(), stripedObservationCounts.length)]; + int stripe = stripeIndex(Thread.currentThread().getId(), stripedObservationCounts.length); + AtomicLong counter = stripedObservationCounts[stripe]; long count = counter.incrementAndGet(); // The active bit is the exact handoff decision. An observation either increments its stripe // before the collector's getAndAdd(BUFFER_ACTIVE_BIT) and takes the direct path, or sees the - // active bit and is buffered in the current generation. + // active bit and may be buffered. The stripe ticket below also checks that it was not counted + // by a later collection that started before this thread read activeGeneration. if ((count & BUFFER_ACTIVE_BIT) == 0) { return false; } + // Allow tests to pause between allocating an observation ticket and reading the generation. + beforeAppendLock.run(); Generation generation = activeGeneration; if (generation == null) { return false; } - beforeAppendLock.run(); appendLock.lock(); try { Generation current = activeGeneration; if (current != generation || !generation.active) { return false; } + if ((count & ~BUFFER_ACTIVE_BIT) <= generationStartCounts[stripe]) { + // This observation incremented its stripe in an earlier generation. The current collector + // already includes it in expectedCount, so buffering it here would make the collector wait + // for an observation that is only replayed after that same wait finishes. + return false; + } while (generation.size >= maxBufferSize && generation.active) { try { bufferSpaceAvailable.await(); @@ -179,8 +191,10 @@ T run( try { activeGeneration = generation; long total = 0; - for (AtomicLong counter : stripedObservationCounts) { - total += counter.getAndAdd(BUFFER_ACTIVE_BIT); + for (int i = 0; i < stripedObservationCounts.length; i++) { + long count = stripedObservationCounts[i].getAndAdd(BUFFER_ACTIVE_BIT); + generationStartCounts[i] = count; + total += count; } expectedCount = total - observationCountOffset; } finally { diff --git a/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java index 3093110d3d..85a34a65bd 100644 --- a/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java +++ b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java @@ -8,6 +8,9 @@ import java.util.ArrayList; import java.util.List; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; @@ -180,83 +183,123 @@ void interruptedAppenderLeavesBoundedBufferWait() throws InterruptedException { } @Test - void lateAppenderCannotBeAddedToTheNextGeneration() throws InterruptedException { - CountDownLatch firstRunStarted = new CountDownLatch(1); - CountDownLatch firstRunMayFinish = new CountDownLatch(1); - CountDownLatch stalled = new CountDownLatch(1); - CountDownLatch release = new CountDownLatch(1); + void lateAppenderCountedByNextGenerationMustNotBeBufferedAgain() throws Exception { + assertLateAppenderHandoff(false); + } + + @Test + void lateAppenderHandoffUsesAbsoluteStripeCountsAfterReset() throws Exception { + assertLateAppenderHandoff(true); + } + + private static void assertLateAppenderHandoff(boolean reset) throws Exception { + CountDownLatch firstSnapshotStarted = new CountDownLatch(1); + CountDownLatch finishFirstSnapshot = new CountDownLatch(1); + CountDownLatch observationCounted = new CountDownLatch(1); + CountDownLatch readGeneration = new CountDownLatch(1); CountDownLatch secondRunStarted = new CountDownLatch(1); - AtomicBoolean appended = new AtomicBoolean(); AtomicLong completedObservations = new AtomicLong(); + AtomicLong secondExpectedCount = new AtomicLong(); + AtomicBoolean pauseFirstAppender = new AtomicBoolean(true); Buffer buffer = new Buffer( - TimeUnit.SECONDS.toNanos(1), + TimeUnit.SECONDS.toNanos(5), 16, () -> { - stalled.countDown(); - try { - release.await(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); + if (pauseFirstAppender.compareAndSet(true, false)) { + observationCounted.countDown(); + awaitLatch(readGeneration); } }); - Thread firstRun = - new Thread( - () -> - buffer.run( - ignored -> { - firstRunStarted.countDown(); - return firstRunMayFinish.getCount() == 0; - }, - () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), - ignored -> {}), - "buffer-first-runner"); - firstRun.setDaemon(true); - firstRun.start(); - assertThat(firstRunStarted.await(5, TimeUnit.SECONDS)).isTrue(); + if (reset) { + assertThat(buffer.append(1.0)).isFalse(); + buffer.observeDirect(completedObservations::incrementAndGet); + } + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future firstRun = + executor.submit( + () -> + buffer.run( + expectedCount -> completedObservations.get() == expectedCount, + () -> { + firstSnapshotStarted.countDown(); + awaitLatch(finishFirstSnapshot); + CounterSnapshot.CounterDataPointSnapshot snapshot = + new CounterSnapshot.CounterDataPointSnapshot( + completedObservations.get(), Labels.EMPTY, null, 0); + if (reset) { + completedObservations.set(0); + buffer.reset(); + } + return snapshot; + }, + ignored -> completedObservations.incrementAndGet())); + awaitLatch(firstSnapshotStarted); - Thread appender = - new Thread( - () -> { - appended.set(buffer.append(1.0)); - if (!appended.get()) { - buffer.observeDirect( - () -> { - completedObservations.incrementAndGet(); - return null; - }); - } - }, - "buffer-late-appender"); - appender.setDaemon(true); - appender.start(); - assertThat(stalled.await(5, TimeUnit.SECONDS)).isTrue(); + // Increment while generation A is active, but do not read activeGeneration yet. + Future appender = + executor.submit( + () -> { + boolean appended = buffer.append(1.0); + if (!appended) { + buffer.observeDirect(completedObservations::incrementAndGet); + } + return appended; + }); + awaitLatch(observationCounted); + finishFirstSnapshot.countDown(); + assertThat(firstRun.get(10, TimeUnit.SECONDS).getValue()).isEqualTo(reset ? 1 : 0); - firstRunMayFinish.countDown(); - firstRun.join(5_000); - assertThat(firstRun.isAlive()).isFalse(); + Future secondRun = + executor.submit( + () -> + buffer.run( + expectedCount -> { + secondExpectedCount.set(expectedCount); + secondRunStarted.countDown(); + return completedObservations.get() == expectedCount; + }, + () -> + new CounterSnapshot.CounterDataPointSnapshot( + completedObservations.get(), Labels.EMPTY, null, 0), + ignored -> completedObservations.incrementAndGet())); + awaitLatch(secondRunStarted); + assertThat(secondExpectedCount).hasValue(1); + // An observation arriving after B's activation still belongs in B's buffer. It must not + // appear in B's snapshot and must be replayed exactly once before the following collection. + assertThat(buffer.append(1.0)).isTrue(); - Thread secondRun = - new Thread( - () -> - buffer.run( - expectedCount -> { - secondRunStarted.countDown(); - return completedObservations.get() == expectedCount; - }, - () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), - ignored -> {}), - "buffer-second-runner"); - secondRun.setDaemon(true); - secondRun.start(); - assertThat(secondRunStarted.await(5, TimeUnit.SECONDS)).isTrue(); - release.countDown(); - appender.join(5_000); - secondRun.join(5_000); + // B includes the paused observation in expectedCount. Buffering it in B would make B wait + // until its own timeout/replay; it must instead complete via the direct observation path. + readGeneration.countDown(); + assertThat(secondRun.get(10, TimeUnit.SECONDS).getValue()).isEqualTo(1); + assertThat(appender.get(10, TimeUnit.SECONDS)).isFalse(); + assertThat(completedObservations).hasValue(2); + assertThat( + buffer + .run( + expectedCount -> completedObservations.get() == expectedCount, + () -> + new CounterSnapshot.CounterDataPointSnapshot( + completedObservations.get(), Labels.EMPTY, null, 0), + ignored -> completedObservations.incrementAndGet()) + .getValue()) + .isEqualTo(2); + } finally { + finishFirstSnapshot.countDown(); + readGeneration.countDown(); + executor.shutdownNow(); + assertThat(executor.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + } + } - assertThat(appender.isAlive()).isFalse(); - assertThat(secondRun.isAlive()).isFalse(); - assertThat(appended).isFalse(); - assertThat(completedObservations).hasValue(1); + private static void awaitLatch(CountDownLatch latch) { + try { + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } } }