diff --git a/docs/apidiffs/current_vs_latest/prometheus-metrics-core.txt b/docs/apidiffs/current_vs_latest/prometheus-metrics-core.txt index ffb4a1d52..136f7f6f1 100644 --- a/docs/apidiffs/current_vs_latest/prometheus-metrics-core.txt +++ b/docs/apidiffs/current_vs_latest/prometheus-metrics-core.txt @@ -1,4 +1,6 @@ Comparing source compatibility of prometheus-metrics-core-1.8.1-SNAPSHOT.jar against prometheus-metrics-core-1.8.0.jar *** MODIFIED CLASS: PUBLIC io.prometheus.metrics.core.exemplars.ExemplarSampler (not serializable) === CLASS FILE FORMAT VERSION: 52.0 <- 52.0 +*** MODIFIED CLASS: PUBLIC io.prometheus.metrics.core.metrics.Histogram$DataPoint (not serializable) + === CLASS FILE FORMAT VERSION: 52.0 <- 52.0 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 c2017995e..6dc68f8e6 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 @@ -1,56 +1,132 @@ package io.prometheus.metrics.core.metrics; +import static java.util.Objects.requireNonNull; + import io.prometheus.metrics.model.snapshots.DataPointSnapshot; import java.util.Arrays; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; +import javax.annotation.Nullable; /** - * Metrics support concurrent write and scrape operations. + * Coordinates concurrent metric observations with collection. + * + *

Collection activates a generation. Observations that start after activation are appended to + * that generation while the collector waits for observations from the previous phase to finish. The + * collector then creates a snapshot, deactivates the generation, and replays its buffered + * observations into the live metric state. * - *

This is implemented by switching to a Buffer when the scrape starts, and applying the values - * from the buffer after the scrape ends. + *

The default collection wait is five seconds. A generation is capped at one million buffered + * observations (about eight MiB of double storage) to keep a stalled collection from growing + * without bound; the cap applies backpressure rather than dropping observations. */ class Buffer { + private static final long BUFFER_ACTIVE_BIT = 1L << 63; + private static final double[] EMPTY_BUFFER = new double[0]; + + // Keep collection bounded without failing healthy scrapes during short periods of scheduler or + // CI-host contention. The one-million-observation cap uses at most 8 MiB for one generation; + // it is deliberately an internal safeguard rather than a data-loss policy. + private static final long DEFAULT_MAX_SPIN_WAIT_NANOS = TimeUnit.SECONDS.toNanos(5); + private static final int DEFAULT_MAX_BUFFER_SIZE = 1_000_000; + private static final int INITIAL_BUFFER_SIZE = 128; + + /** Observations buffered during one collection cycle. */ + private static final class Generation { + private double[] values = EMPTY_BUFFER; + private int size; + private boolean active = true; + } - private static final long bufferActiveBit = 1L << 63; // Tracking observation counts requires an AtomicLong for coordination between recording and // collecting. AtomicLong does much worse under contention than the LongAdder instances used - // elsewhere to hold aggregated state. To improve, we stripe the AtomicLong into N instances, - // where N is the number of available processors. Each record operation chooses the appropriate - // instance to use based on the modulo of its thread id and N. This is a more naive / simple - // implementation compared to the striping used under the hood in java.util.concurrent classes - // like LongAdder - contention and hot spots can still occur if recording thread ids happen to - // resolve to the same index. Further improvement is possible. + // elsewhere to hold aggregated state. To reduce contention, the count is striped across the + // 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; - private double[] observationBuffer = new double[0]; - private int bufferPos = 0; - private boolean reset = false; - + private final ReentrantLock observationLock = new ReentrantLock(); + private boolean reset; + private long observationCountOffset; + @Nullable private volatile Generation activeGeneration; ReentrantLock appendLock = new ReentrantLock(); ReentrantLock runLock = new ReentrantLock(); - Condition bufferFilled = appendLock.newCondition(); + private final Condition bufferSpaceAvailable = appendLock.newCondition(); + private final long maxSpinWaitNanos; + private final int maxBufferSize; + private final Runnable beforeAppendLock; Buffer() { + this(DEFAULT_MAX_SPIN_WAIT_NANOS, DEFAULT_MAX_BUFFER_SIZE, () -> {}); + } + + Buffer(long maxSpinWaitNanos) { + this(maxSpinWaitNanos, DEFAULT_MAX_BUFFER_SIZE, () -> {}); + } + + Buffer(long maxSpinWaitNanos, int maxBufferSize, Runnable beforeAppendLock) { + if (maxBufferSize <= 0) { + throw new IllegalArgumentException("maxBufferSize must be positive"); + } + this.maxSpinWaitNanos = maxSpinWaitNanos; + this.maxBufferSize = maxBufferSize; + this.beforeAppendLock = beforeAppendLock; stripedObservationCounts = new AtomicLong[Runtime.getRuntime().availableProcessors()]; for (int i = 0; i < stripedObservationCounts.length; i++) { - stripedObservationCounts[i] = new AtomicLong(0); + stripedObservationCounts[i] = new AtomicLong(); } } boolean append(double value) { - int index = stripeIndex(Thread.currentThread().getId(), stripedObservationCounts.length); - AtomicLong observationCountForThread = stripedObservationCounts[index]; - long count = observationCountForThread.incrementAndGet(); - if ((count & bufferActiveBit) == 0) { - return false; // sign bit not set -> buffer not active. - } else { - doAppend(value); + AtomicLong counter = + stripedObservationCounts[ + stripeIndex(Thread.currentThread().getId(), stripedObservationCounts.length)]; + 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. + if ((count & BUFFER_ACTIVE_BIT) == 0) { + return false; + } + Generation generation = activeGeneration; + if (generation == null) { + return false; + } + beforeAppendLock.run(); + appendLock.lock(); + try { + Generation current = activeGeneration; + if (current != generation || !generation.active) { + return false; + } + while (generation.size >= maxBufferSize && generation.active) { + try { + bufferSpaceAvailable.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } + } + if (!generation.active) { + return false; + } + if (generation.size >= generation.values.length) { + int doubled = + generation.values.length > maxBufferSize / 2 + ? maxBufferSize + : generation.values.length * 2; + generation.values = + Arrays.copyOf( + generation.values, Math.min(maxBufferSize, Math.max(INITIAL_BUFFER_SIZE, doubled))); + } + generation.values[generation.size++] = value; return true; + } finally { + appendLock.unlock(); } } @@ -58,89 +134,105 @@ static int stripeIndex(long threadId, int stripeCount) { return (int) Math.floorMod(threadId, stripeCount); } - private void doAppend(double amount) { - appendLock.lock(); - try { - if (bufferPos >= observationBuffer.length) { - observationBuffer = Arrays.copyOf(observationBuffer, observationBuffer.length + 128); - } - observationBuffer[bufferPos] = amount; - bufferPos++; + void reset() { + reset = true; + } - bufferFilled.signalAll(); + T observeDirect(Supplier observeFunction) { + // In steady state this is the lock-free path used before this buffer was introduced. Keep the + // lock only while a generation is active, so direct observations cannot race collection/replay. + if (activeGeneration == null) { + return observeFunction.get(); + } + observationLock.lock(); + try { + return observeFunction.get(); } finally { - appendLock.unlock(); + observationLock.unlock(); } } - /** Must be called by the runnable in the run() method. */ - void reset() { - reset = true; + @SuppressWarnings("ThreadPriorityCheck") + T run( + Function complete, + Supplier createResult, + Consumer observeFunction) { + return requireNonNull(run(complete, createResult, observeFunction, true)); } @SuppressWarnings("ThreadPriorityCheck") + @Nullable T run( Function complete, Supplier createResult, - Consumer observeFunction) { + Consumer observeFunction, + boolean failOnTimeout) { + Generation generation = new Generation(); double[] buffer; int bufferSize; - T result; - + boolean timedOut = false; + T result = null; runLock.lock(); try { - // Signal that the buffer is active. - long expectedCount = 0L; - for (AtomicLong observationCount : stripedObservationCounts) { - expectedCount += observationCount.getAndAdd(bufferActiveBit); + long expectedCount; + appendLock.lock(); + try { + activeGeneration = generation; + long total = 0; + for (AtomicLong counter : stripedObservationCounts) { + total += counter.getAndAdd(BUFFER_ACTIVE_BIT); + } + expectedCount = total - observationCountOffset; + } finally { + appendLock.unlock(); } - + long deadline = System.nanoTime() + maxSpinWaitNanos; while (!complete.apply(expectedCount)) { - // Wait until all in-flight threads have added their observations to the histogram / - // summary. - // we can't use a condition here, because the other thread doesn't have a lock as it's on - // the fast path. - Thread.yield(); - } - result = createResult.get(); - - // Signal that the buffer is inactive. - long expectedBufferSize = 0; - if (reset) { - for (AtomicLong observationCount : stripedObservationCounts) { - expectedBufferSize += observationCount.getAndSet(0) & ~bufferActiveBit; - } - reset = false; - } else { - for (AtomicLong observationCount : stripedObservationCounts) { - expectedBufferSize += observationCount.addAndGet(bufferActiveBit); + if (System.nanoTime() - deadline >= 0) { + timedOut = true; + break; } + Thread.yield(); } - expectedBufferSize -= expectedCount; - - appendLock.lock(); + observationLock.lock(); try { - while (bufferPos < expectedBufferSize) { - // Wait until all in-flight threads have added their observations to the buffer. - bufferFilled.await(); - } + result = timedOut ? null : createResult.get(); } finally { - appendLock.unlock(); + try { + appendLock.lock(); + try { + generation.active = false; + for (AtomicLong counter : stripedObservationCounts) { + counter.addAndGet(BUFFER_ACTIVE_BIT); + } + if (reset) { + observationCountOffset += expectedCount; + reset = false; + } + buffer = generation.values; + bufferSize = generation.size; + generation.values = EMPTY_BUFFER; + generation.size = 0; + bufferSpaceAvailable.signalAll(); + } finally { + appendLock.unlock(); + } + for (int i = 0; i < bufferSize; i++) { + observeFunction.accept(buffer[i]); + } + // Keep the inactive generation visible until replay completes. An appender that loses the + // generation race must take observationLock before observing directly. + activeGeneration = null; + } finally { + observationLock.unlock(); + } } - - buffer = observationBuffer; - bufferSize = bufferPos; - observationBuffer = new double[0]; - bufferPos = 0; - } catch (InterruptedException e) { - throw new RuntimeException(e); + if (timedOut && failOnTimeout) { + throw new IllegalStateException("Timed out while waiting for in-flight observations."); + } + return result; } finally { runLock.unlock(); } - - for (int i = 0; i < bufferSize; i++) { - observeFunction.accept(buffer[i]); - } - return result; } } diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Histogram.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Histogram.java index c4bb1f5fe..d03ac9b97 100644 --- a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Histogram.java +++ b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Histogram.java @@ -243,7 +243,8 @@ public void observe(double value) { return; } if (!buffer.append(value)) { - doObserve(value, false); + boolean nativeBucketCreated = buffer.observeDirect(() -> doObserve(value)); + maybeResetOrScaleDown(value, nativeBucketCreated); } if (exemplarSampler != null) { exemplarSampler.observe(value); @@ -257,14 +258,15 @@ public void observeWithExemplar(double value, Labels labels) { return; } if (!buffer.append(value)) { - doObserve(value, false); + boolean nativeBucketCreated = buffer.observeDirect(() -> doObserve(value)); + maybeResetOrScaleDown(value, nativeBucketCreated); } if (exemplarSampler != null) { exemplarSampler.observeWithExemplar(value, labels); } } - private void doObserve(double value, boolean fromBuffer) { + private boolean doObserve(double value) { // classicUpperBounds is an empty array if this is a native histogram only. for (int i = 0; i < classicUpperBounds.length; ++i) { // The last bucket is +Inf, so we always increment. @@ -287,16 +289,7 @@ private void doObserve(double value, boolean fromBuffer) { count .increment(); // must be the last step, because count is used to signal that the operation // is complete. - if (!fromBuffer) { - // maybeResetOrScaleDown will switch to the buffer, - // which won't work if we are currently still processing observations from the buffer. - // The reason is that before switching to the buffer we wait for all pending observations to - // be counted. - // If we do this while still applying observations from the buffer, the pending observations - // from - // the buffer will never be counted, and the buffer.run() method will wait forever. - maybeResetOrScaleDown(value, nativeBucketCreated); - } + return nativeBucketCreated; } private HistogramSnapshot.HistogramDataPointSnapshot collect(Labels labels) { @@ -339,7 +332,7 @@ private HistogramSnapshot.HistogramDataPointSnapshot collect(Labels labels) { createdTimeMillis); } }, - v -> doObserve(v, true)); + this::doObserve); } private boolean addToNativeBucket(double value, ConcurrentHashMap buckets) { @@ -444,7 +437,8 @@ private void maybeResetOrScaleDown(double value, boolean nativeBucketCreated) { } return null; }, - v -> doObserve(v, true)); + this::doObserve, + false); } else if (nativeBucketCreated) { // If a new bucket was created we need to check if nativeMaxBuckets is exceeded // and scale down if so. @@ -453,7 +447,11 @@ private void maybeResetOrScaleDown(double value, boolean nativeBucketCreated) { if (wasReset.get()) { // We just discarded the newly observed value. Observe it again. if (!buffer.append(value)) { - doObserve(value, true); + buffer.observeDirect( + () -> { + doObserve(value); + return null; + }); } } } @@ -488,7 +486,8 @@ private void maybeScaleDown(AtomicBoolean wasReset) { doubleBucketWidth(); return null; }, - v -> doObserve(v, true)); + this::doObserve, + false); } // maybeReset is called in the synchronized block while new observations go into the buffer. diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Summary.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Summary.java index 043f31129..d75ac6c74 100644 --- a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Summary.java +++ b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Summary.java @@ -185,7 +185,11 @@ public void observe(double value) { return; } if (!buffer.append(value)) { - doObserve(value); + buffer.observeDirect( + () -> { + doObserve(value); + return null; + }); } if (exemplarSampler != null) { exemplarSampler.observe(value); @@ -198,7 +202,11 @@ public void observeWithExemplar(double value, Labels labels) { return; } if (!buffer.append(value)) { - doObserve(value); + buffer.observeDirect( + () -> { + doObserve(value); + return null; + }); } if (exemplarSampler != null) { exemplarSampler.observeWithExemplar(value, labels); 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 dccd3b4eb..3093110d3 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 @@ -1,7 +1,17 @@ package io.prometheus.metrics.core.metrics; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; +import io.prometheus.metrics.model.snapshots.CounterSnapshot; +import io.prometheus.metrics.model.snapshots.Labels; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.Test; class BufferTest { @@ -12,4 +22,241 @@ void stripeIndexDoesNotOverflowWhenThreadIdNarrowsToIntegerMinValue() { assertThat(Buffer.stripeIndex(2_147_483_648L, 6)).isEqualTo(2); assertThat(Buffer.stripeIndex(2_147_483_648L, 12)).isEqualTo(8); } + + @Test + void timeoutDeactivatesBufferAndReplaysBufferedObservations() throws InterruptedException { + Buffer buffer = new Buffer(TimeUnit.SECONDS.toNanos(1)); + CountDownLatch spinWaitStarted = new CountDownLatch(1); + List replayedObservations = new ArrayList<>(); + AtomicBoolean timedOut = new AtomicBoolean(false); + + Thread runner = + new Thread( + () -> { + try { + buffer.run( + expectedCount -> { + spinWaitStarted.countDown(); + return false; + }, + () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), + replayedObservations::add); + } catch (IllegalStateException expected) { + timedOut.set(true); + } + }, + "buffer-timeout-runner"); + runner.setDaemon(true); + runner.start(); + + assertThat(spinWaitStarted.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(buffer.append(1.0)).isTrue(); + runner.join(5_000); + + assertThat(timedOut).isTrue(); + assertThat(replayedObservations).containsExactly(1.0); + assertThat(buffer.append(2.0)).isFalse(); + } + + @Test + void timeoutDoesNotCreateSnapshot() { + Buffer buffer = new Buffer(TimeUnit.MILLISECONDS.toNanos(1)); + + assertThatExceptionOfType(IllegalStateException.class) + .isThrownBy( + () -> + buffer.run( + expectedCount -> false, + () -> { + throw new AssertionError("snapshot should not be created"); + }, + ignored -> {})) + .withMessage("Timed out while waiting for in-flight observations."); + } + + @Test + void fullBufferUnblocksAppenderWhenGenerationIsDeactivated() throws InterruptedException { + CountDownLatch runStarted = new CountDownLatch(1); + CountDownLatch secondAppenderEntered = new CountDownLatch(1); + AtomicLong beforeAppendCount = new AtomicLong(); + AtomicReference appended = new AtomicReference<>(); + AtomicBoolean timedOut = new AtomicBoolean(); + Buffer buffer = + new Buffer( + TimeUnit.MILLISECONDS.toNanos(250), + 1, + () -> { + if (beforeAppendCount.incrementAndGet() == 2) { + secondAppenderEntered.countDown(); + } + }); + Thread runner = + new Thread( + () -> { + try { + buffer.run( + ignored -> { + runStarted.countDown(); + return false; + }, + () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), + ignored -> {}); + } catch (IllegalStateException expected) { + timedOut.set(true); + } + }, + "buffer-full-runner"); + runner.setDaemon(true); + runner.start(); + assertThat(runStarted.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(buffer.append(1.0)).isTrue(); + + Thread appender = new Thread(() -> appended.set(buffer.append(2.0)), "buffer-full-appender"); + appender.setDaemon(true); + appender.start(); + assertThat(secondAppenderEntered.await(5, TimeUnit.SECONDS)).isTrue(); + runner.join(5_000); + appender.join(5_000); + + assertThat(timedOut).isTrue(); + assertThat(appender.isAlive()).isFalse(); + assertThat(appended).hasValue(false); + } + + @Test + void interruptedAppenderLeavesBoundedBufferWait() throws InterruptedException { + CountDownLatch runStarted = new CountDownLatch(1); + CountDownLatch secondAppenderEntered = new CountDownLatch(1); + AtomicLong beforeAppendCount = new AtomicLong(); + AtomicBoolean interrupted = new AtomicBoolean(); + AtomicReference appended = new AtomicReference<>(); + Buffer buffer = + new Buffer( + TimeUnit.SECONDS.toNanos(1), + 1, + () -> { + if (beforeAppendCount.incrementAndGet() == 2) { + secondAppenderEntered.countDown(); + } + }); + Thread runner = + new Thread( + () -> { + try { + buffer.run( + ignored -> { + runStarted.countDown(); + return false; + }, + () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), + ignored -> {}); + } catch (IllegalStateException expected) { + // The runner is only used to hold the generation open for this test. + } + }, + "buffer-interrupt-runner"); + runner.setDaemon(true); + runner.start(); + assertThat(runStarted.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(buffer.append(1.0)).isTrue(); + + Thread appender = + new Thread( + () -> { + appended.set(buffer.append(2.0)); + interrupted.set(Thread.currentThread().isInterrupted()); + }, + "buffer-interrupt-appender"); + appender.setDaemon(true); + appender.start(); + assertThat(secondAppenderEntered.await(5, TimeUnit.SECONDS)).isTrue(); + appender.interrupt(); + appender.join(5_000); + runner.join(5_000); + + assertThat(appender.isAlive()).isFalse(); + assertThat(appended).hasValue(false); + assertThat(interrupted).isTrue(); + } + + @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); + CountDownLatch secondRunStarted = new CountDownLatch(1); + AtomicBoolean appended = new AtomicBoolean(); + AtomicLong completedObservations = new AtomicLong(); + Buffer buffer = + new Buffer( + TimeUnit.SECONDS.toNanos(1), + 16, + () -> { + stalled.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + 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(); + + 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(); + + firstRunMayFinish.countDown(); + firstRun.join(5_000); + assertThat(firstRun.isAlive()).isFalse(); + + 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); + + assertThat(appender.isAlive()).isFalse(); + assertThat(secondRun.isAlive()).isFalse(); + assertThat(appended).isFalse(); + assertThat(completedObservations).hasValue(1); + } }