You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/07/23 08:28:50 UTC
(pulsar) branch branch-4.0 updated: [improve][broker][branch-4.0] Trace the asynchronous tasks in logs when loading topics (#26163) (#26224)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new d7c265988780 [improve][broker][branch-4.0] Trace the asynchronous tasks in logs when loading topics (#26163) (#26224)
d7c265988780 is described below
commit d7c2659887806d8619353acf0761b85f71f3f39e
Author: Yunze Xu <xy...@163.com>
AuthorDate: Thu Jul 23 16:14:43 2026 +0800
[improve][broker][branch-4.0] Trace the asynchronous tasks in logs when loading topics (#26163) (#26224)
---
.../pulsar/broker/service/BrokerService.java | 109 +++++++---------
.../pulsar/broker/service/TopicLoadingContext.java | 34 ++---
.../apache/pulsar/common/util/LatencyTracer.java | 110 ++++++++++++++++
.../pulsar/common/util/LatencyTracerTest.java | 143 +++++++++++++++++++++
4 files changed, 313 insertions(+), 83 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index c759ff2c4351..e13c11e3dde7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -1119,32 +1119,34 @@ public class BrokerService implements Closeable {
return FutureUtil.failedFuture(new NotAllowedException(
"Broker is unable to load persistent topic"));
}
+ final var timeoutSeconds = pulsar.getConfiguration().getTopicLoadTimeoutSeconds();
final CompletableFuture<Optional<Topic>> topicFuture = FutureUtil.createFutureWithTimeout(
- Duration.ofSeconds(pulsar.getConfiguration().getTopicLoadTimeoutSeconds()), executor(),
+ Duration.ofSeconds(timeoutSeconds), executor(),
() -> FAILED_TO_LOAD_TOPIC_TIMEOUT_EXCEPTION);
final var context = new TopicLoadingContext(topicName, createIfMissing, topicFuture);
if (properties != null) {
context.setProperties(properties);
}
topicFuture.exceptionally(t -> {
- final var now = System.nanoTime();
+ final var latency = context.traceAndGetLatency("fail").description();
if (FutureUtil.unwrapCompletionException(t) instanceof TimeoutException) {
- log.warn("Failed to load {} after {} ms", topicName, context.latencyMs(now));
+ log.warn("Failed to load {} within {} s (latency: {})", topicName, timeoutSeconds, latency);
} else {
- log.warn("Failed to load {} after {} ms", topicName, context.latencyString(now), t);
+ log.warn("Failed to load {} (latency: {})", topicName, latency, t);
}
pulsarStats.recordTopicLoadFailed();
return Optional.empty();
});
- checkNonPartitionedTopicExists(topicName).thenAccept(exists -> {
+ context.trace("topic exists", checkNonPartitionedTopicExists(topicName)).thenAccept(exists -> {
if (!exists && !createIfMissing) {
topicFuture.complete(Optional.empty());
return;
}
// The topic level policies are not needed now, but the meaning of calling
// "getTopicPoliciesBypassSystemTopic" will wait for system topic policies initialization.
- getTopicPoliciesBypassSystemTopic(topicName, TopicPoliciesService.GetType.LOCAL_ONLY)
- .thenRun(() -> {
+ final var systemTopicLoadFuture = context.trace("system topic",
+ getTopicPoliciesBypassSystemTopic(topicName, TopicPoliciesService.GetType.LOCAL_ONLY));
+ systemTopicLoadFuture.thenRun(() -> {
final var inserted = new MutableBoolean(false);
final var cachedFuture = topics.computeIfAbsent(topicName.toString(), ___ -> {
inserted.setTrue();
@@ -1155,7 +1157,7 @@ public class BrokerService implements Closeable {
// checked if the `topics` cache includes this topic before, so the latency is not the
// actual loading latency that should not be recorded in metrics.
log.info("[{}] Finished loading from other concurrent loading task (latency: {})",
- topicName, context.latencyString(System.nanoTime()));
+ topicName, context.getLatency().description());
cachedFuture.whenComplete((optTopic, e) -> {
if (e == null) {
topicFuture.complete(optTopic);
@@ -1165,18 +1167,13 @@ public class BrokerService implements Closeable {
});
}
}).exceptionally(e -> {
- pulsar.getExecutor().execute(() -> topics.remove(topicName.toString(), topicFuture));
- final Throwable rc = FutureUtil.unwrapCompletionException(e);
- final String errorInfo = String.format("Topic creation encountered an exception by initialize"
- + " topic policies service. topic_name=%s error_message=%s", topicName,
- rc.getMessage());
- log.error(errorInfo, rc);
- topicFuture.completeExceptionally(rc);
+ log.warn("[{}] Topic creation encountered an exception"
+ + " by initialize topic policies service", topicName);
+ failTopicFuture(topicName.toString(), topicFuture, e);
return null;
});
}).exceptionally(e -> {
- pulsar.getExecutor().execute(() -> topics.remove(topicName.toString(), topicFuture));
- topicFuture.completeExceptionally(e.getCause());
+ failTopicFuture(topicName.toString(), topicFuture, e);
return null;
});
return topicFuture;
@@ -1247,6 +1244,14 @@ public class BrokerService implements Closeable {
}
}
+ private void failTopicFuture(String topic, CompletableFuture<Optional<Topic>> topicFuture, Throwable throwable) {
+ pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
+ final Throwable rc = FutureUtil.unwrapCompletionException(throwable);
+ // It will trigger the logging for exception and traced latencies in topicFuture's exceptionally callback, so
+ // we don't need to add an extra log before it.
+ topicFuture.completeExceptionally(rc);
+ }
+
private CompletableFuture<Optional<TopicPolicies>> getTopicPoliciesBypassSystemTopic(@NonNull TopicName topicName,
TopicPoliciesService.GetType type) {
if (ExtensibleLoadManagerImpl.isInternalTopic(topicName.toString())) {
@@ -1706,7 +1711,7 @@ public class BrokerService implements Closeable {
ownedFuture = checkTopicNsOwnership(topic);
}
final var topicFuture = context.getTopicFuture();
- ownedFuture
+ context.trace("ownership", ownedFuture)
.thenRun(() -> {
final Semaphore topicLoadSemaphore = topicLoadRequestSemaphore.get();
@@ -1718,8 +1723,7 @@ public class BrokerService implements Closeable {
// do not recreate topic if topic is already migrated and deleted by broker
// so, avoid creating a new topic if migration is already started
if (ex != null && (ex.getCause() instanceof TopicMigratedException)) {
- pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
- topicFuture.completeExceptionally(ex.getCause());
+ failTopicFuture(topic, topicFuture, ex);
return null;
}
createPendingLoadTopic();
@@ -1732,8 +1736,7 @@ public class BrokerService implements Closeable {
}
}
}).exceptionally(ex -> {
- pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
- topicFuture.completeExceptionally(ex.getCause());
+ failTopicFuture(topic, topicFuture, ex);
return null;
});
@@ -1773,7 +1776,7 @@ public class BrokerService implements Closeable {
TopicName topicName = context.getTopicName();
final var topic = topicName.toString();
final var topicFuture = context.getTopicFuture();
- checkTopicNsOwnership(topic).thenRun(() -> {
+ context.trace("2nd ownership", checkTopicNsOwnership(topic)).thenRun(() -> {
CompletableFuture<Map<String, String>> propertiesFuture;
if (context.getProperties() == null) {
//Read properties from storage when loading topic.
@@ -1781,19 +1784,16 @@ public class BrokerService implements Closeable {
} else {
propertiesFuture = CompletableFuture.completedFuture(context.getProperties());
}
- propertiesFuture.thenAccept(finalProperties -> {
+ context.trace("properties", propertiesFuture).thenAccept(finalProperties -> {
context.setProperties(finalProperties);
//TODO add topicName in properties?
createPersistentTopic0(context);
}).exceptionally(throwable -> {
- log.warn("[{}] Read topic property failed", topic, throwable);
- pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
- topicFuture.completeExceptionally(throwable);
+ failTopicFuture(topic, topicFuture, throwable);
return null;
});
}).exceptionally(e -> {
- pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
- topicFuture.completeExceptionally(e.getCause());
+ failTopicFuture(topic, topicFuture, e);
return null;
});
}
@@ -1806,10 +1806,8 @@ public class BrokerService implements Closeable {
final var createIfMissing = context.isCreateIfMissing();
if (isTransactionInternalName(topicName)) {
- String msg = String.format("Can not create transaction system topic %s", topic);
- log.warn(msg);
- pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
- topicFuture.completeExceptionally(new NotAllowedException(msg));
+ failTopicFuture(topic, topicFuture, new NotAllowedException("Can not create transaction system topic "
+ + topic));
return;
}
@@ -1826,6 +1824,7 @@ public class BrokerService implements Closeable {
n.recycle();
return found;
}), (managedLedgerConfig, exists) -> {
+ context.trace("ml-config");
if (isBrokerEntryMetadataEnabled() || isBrokerPayloadProcessorEnabled()) {
// init managedLedger interceptor
Set<BrokerEntryMetadataInterceptor> interceptors = new HashSet<>();
@@ -1877,25 +1876,24 @@ public class BrokerService implements Closeable {
@Override
public void openLedgerComplete(ManagedLedger ledger, Object ctx) {
try {
+ context.trace("open-ml");
PersistentTopic persistentTopic = isSystemTopic(topic)
? new SystemTopic(topic, ledger, BrokerService.this)
: newTopic(topic, ledger, BrokerService.this, PersistentTopic.class);
persistentTopic.setCreateFuture(topicFuture);
- persistentTopic
- .initialize()
- .thenCompose(__ -> persistentTopic.preCreateSubscriptionForCompactionIfNeeded())
- .thenCompose(__ -> persistentTopic.checkReplication())
- .thenCompose(v -> {
- // Also check dedup status
- return persistentTopic.checkDeduplicationStatus();
- })
+ context.trace("init", persistentTopic.initialize())
+ .thenCompose(__ -> context.trace("pre-create compacted sub",
+ persistentTopic.preCreateSubscriptionForCompactionIfNeeded()))
+ .thenCompose(__ -> context.trace("replication",
+ persistentTopic.checkReplication()))
+ .thenCompose(v -> context.trace("deduplication",
+ persistentTopic.checkDeduplicationStatus()))
.thenRun(() -> {
- long nowInNanos = System.nanoTime();
- long topicLoadLatencyMs = context.latencyMs(nowInNanos);
+ final var latency = context.traceAndGetLatency("done");
log.info("Created topic {} - dedup is {} (latency: {})", topic,
persistentTopic.isDeduplicationEnabled() ? "enabled" : "disabled",
- context.latencyString(nowInNanos));
- pulsarStats.recordTopicLoadTimeValue(topic, topicLoadLatencyMs);
+ latency.description());
+ pulsarStats.recordTopicLoadTimeValue(topic, latency.elapsedInMillis());
if (!topicFuture.complete(Optional.of(persistentTopic))) {
// Check create persistent topic timeout.
if (topicFuture.isCompletedExceptionally()) {
@@ -1935,9 +1933,7 @@ public class BrokerService implements Closeable {
return null;
});
} catch (Exception e) {
- log.warn("Failed to create topic {}: {}", topic, e.getMessage());
- pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
- topicFuture.completeExceptionally(e);
+ failTopicFuture(topic, topicFuture, e);
}
}
@@ -1949,9 +1945,7 @@ public class BrokerService implements Closeable {
loadFuture.completeExceptionally(exception);
topicFuture.complete(Optional.empty());
} else {
- log.warn("Failed to create topic {}", topic, exception);
- pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
- topicFuture.completeExceptionally(new PersistenceException(exception));
+ failTopicFuture(topic, topicFuture, new PersistenceException(exception));
}
}
}, () -> isTopicNsOwnedByBrokerAsync(topicName), null);
@@ -1959,12 +1953,9 @@ public class BrokerService implements Closeable {
}).exceptionally((exception) -> {
boolean migrationFailure = exception.getCause() instanceof TopicMigratedException;
String msg = migrationFailure ? "Topic is already migrated" :
- "Failed to get topic configuration:";
- log.warn("[{}] {} {}", topic, msg, exception.getMessage(), exception);
- // remove topic from topics-map in different thread to avoid possible deadlock if
- // createPersistentTopic-thread only tries to handle this future-result
- pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
- topicFuture.completeExceptionally(exception);
+ "Failed to get topic configuration";
+ log.warn("[{}] {}", topic, msg);
+ failTopicFuture(topic, topicFuture, exception);
return null;
});
}
@@ -3317,9 +3308,9 @@ public class BrokerService implements Closeable {
return;
}
- pendingTopic.polledFromQueue();
+ pendingTopic.trace("queued");
final String topic = pendingTopic.getTopicName().toString();
- checkTopicNsOwnership(topic).thenRun(() -> {
+ pendingTopic.trace("ownership", checkTopicNsOwnership(topic)).thenRun(() -> {
CompletableFuture<Optional<Topic>> pendingFuture = pendingTopic.getTopicFuture();
final Semaphore topicLoadSemaphore = topicLoadRequestSemaphore.get();
final boolean acquiredPermit = topicLoadSemaphore.tryAcquire();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicLoadingContext.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicLoadingContext.java
index 9e3ed230cd26..b9a668f92b26 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicLoadingContext.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicLoadingContext.java
@@ -21,19 +21,15 @@ package org.apache.pulsar.broker.service;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.ConcurrentLinkedQueue;
import lombok.Getter;
-import lombok.RequiredArgsConstructor;
import lombok.Setter;
import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.util.LatencyTracer;
import org.jspecify.annotations.Nullable;
-@RequiredArgsConstructor
-public class TopicLoadingContext {
+public class TopicLoadingContext extends LatencyTracer {
- private static final String EXAMPLE_LATENCY_OUTPUTS = "1234 ms (queued: 567)";
-
- private final long startNs = System.nanoTime();
@Getter
private final TopicName topicName;
@Getter
@@ -43,23 +39,13 @@ public class TopicLoadingContext {
@Getter
@Setter
@Nullable private Map<String, String> properties;
- private long polledFromQueueNs = -1L;
-
- public void polledFromQueue() {
- polledFromQueueNs = System.nanoTime();
- }
-
- public long latencyMs(long nowInNanos) {
- return TimeUnit.NANOSECONDS.toMillis(nowInNanos - startNs);
- }
- public String latencyString(long nowInNanos) {
- final var builder = new StringBuilder(EXAMPLE_LATENCY_OUTPUTS.length());
- builder.append(latencyMs(nowInNanos));
- builder.append(" ms");
- if (polledFromQueueNs >= 0) {
- builder.append(" (queued: ").append(latencyMs(polledFromQueueNs)).append(")");
- }
- return builder.toString();
+ public TopicLoadingContext(TopicName topicName, boolean createIfMissing,
+ CompletableFuture<Optional<Topic>> topicFuture) {
+ // The topic loading could be ended asynchronously by a timeout event, so we need a thread safe queue here
+ super(new ConcurrentLinkedQueue<>(), System::nanoTime);
+ this.topicName = topicName;
+ this.createIfMissing = createIfMissing;
+ this.topicFuture = topicFuture;
}
}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/LatencyTracer.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/LatencyTracer.java
new file mode 100644
index 000000000000..afc335a16e4b
--- /dev/null
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/LatencyTracer.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.util;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.experimental.Accessors;
+
+public class LatencyTracer {
+
+ private final Queue<Timepoint> timepoints;
+ private final NanoTimeSupplier nanoTimeSupplier;
+ private final long startNs;
+
+ public LatencyTracer(Queue<Timepoint> timepoints, NanoTimeSupplier nanoTimeSupplier) {
+ this.timepoints = timepoints;
+ this.nanoTimeSupplier = nanoTimeSupplier;
+ this.startNs = nanoTimeSupplier.getNanos();
+ }
+
+ public <T> CompletableFuture<T> trace(String message, CompletableFuture<T> future) {
+ if (future.isDone()) {
+ return future;
+ }
+ return future.whenComplete((__, ___) -> trace(message));
+ }
+
+ public void trace(String action) {
+ timepoints.add(new Timepoint(action, nanoTimeSupplier.getNanos()));
+ }
+
+ public Snapshot getLatency() {
+ final List<Timepoint> timepoints = new ArrayList<>(this.timepoints);
+ if (timepoints.isEmpty()) {
+ return new Snapshot(startNs, 0, "total: 0 ms");
+ }
+ final StringBuilder sb = new StringBuilder();
+ final long totalLatencyNs = TimeUnit.NANOSECONDS.toMillis(timepoints.get(timepoints.size() - 1).timeInNanos
+ - startNs);
+ sb.append("total: ").append(totalLatencyNs).append(" ms");
+ long prevNs = startNs;
+ for (final Timepoint tp : timepoints) {
+ sb.append(", ").append(tp.name).append(": ");
+ long latencyMs = TimeUnit.NANOSECONDS.toMillis(tp.timeInNanos - prevNs);
+ if (latencyMs > 0) {
+ sb.append(latencyMs).append(" ms");
+ } else {
+ sb.append(TimeUnit.NANOSECONDS.toMicros(tp.timeInNanos - prevNs)).append(" us");
+ }
+ prevNs = tp.timeInNanos;
+ }
+ return new Snapshot(prevNs, totalLatencyNs, sb.toString());
+ }
+
+ public Snapshot traceAndGetLatency(String action) {
+ trace(action);
+ return getLatency();
+ }
+
+ public interface NanoTimeSupplier {
+
+ long getNanos();
+ }
+
+ @Accessors(fluent = true)
+ @Getter
+ @RequiredArgsConstructor
+ public static final class Timepoint {
+
+ private final String name;
+ private final long timeInNanos;
+ }
+
+ /**
+ * A snapshot of the latency tracer at a given moment.
+ * `endTimeInNanos` the latest traced timestamp in nanoseconds
+ * `elapsedInMillis` the elapsed time in milliseconds from when the tracer was created
+ * `description` the detailed description for traced latencies, e.g. "total: 100 ms, A: 60 ms, B: 40 ms"
+ */
+ @Accessors(fluent = true)
+ @Getter
+ @RequiredArgsConstructor
+ public static final class Snapshot {
+
+ private final long endTimeInNanos;
+ private final long elapsedInMillis;
+ private final String description;
+ }
+}
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/LatencyTracerTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/LatencyTracerTest.java
new file mode 100644
index 000000000000..d3d638d096e5
--- /dev/null
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/LatencyTracerTest.java
@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.util;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNotSame;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
+import java.util.LinkedList;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import org.testng.annotations.Test;
+
+@Test(groups = "utils")
+public class LatencyTracerTest {
+
+ private static LatencyTracer.NanoTimeSupplier testNanoTimeSupplier(long... nanoTimes) {
+ return new LatencyTracer.NanoTimeSupplier() {
+ final LinkedList<Long> nanoTimesQueue = new LinkedList<>();
+
+ {
+ for (long nanoTime : nanoTimes) {
+ nanoTimesQueue.add(nanoTime);
+ }
+ }
+
+ @Override
+ public long getNanos() {
+ final Long nanos = nanoTimesQueue.poll();
+ assertNotNull(nanos);
+ return nanos;
+ }
+ };
+ }
+
+ @Test
+ public void testMulti() {
+ LatencyTracer tracer = new LatencyTracer(new LinkedList<>(), testNanoTimeSupplier(10_000_000L, 30_000_000L,
+ 70_000_000L, 80_000_000L));
+ tracer.trace("A");
+ tracer.trace("B");
+ LatencyTracer.Snapshot snapshot = tracer.getLatency();
+ assertEquals(snapshot.description(), "total: 60 ms, A: 20 ms, B: 40 ms");
+ assertEquals(snapshot.elapsedInMillis(), 60);
+
+ tracer.trace("C");
+ snapshot = tracer.getLatency();
+ assertEquals(snapshot.description(), "total: 70 ms, A: 20 ms, B: 40 ms, C: 10 ms");
+ assertEquals(snapshot.elapsedInMillis(), 70);
+ }
+
+ @Test
+ public void testEmpty() {
+ LatencyTracer tracer = new LatencyTracer(new LinkedList<>(), testNanoTimeSupplier(0L, 20_000_000L));
+ LatencyTracer.Snapshot snapshot = tracer.getLatency();
+ assertEquals(snapshot.description(), "total: 0 ms");
+ assertEquals(snapshot.elapsedInMillis(), 0);
+ }
+
+ @Test
+ public void testZeroMs() {
+ LatencyTracer tracer = new LatencyTracer(new LinkedList<>(), testNanoTimeSupplier(0L, 999_999L, 2_000_000L,
+ 2_100_000L));
+ tracer.trace("A");
+ tracer.trace("B");
+ tracer.trace("C");
+ LatencyTracer.Snapshot snapshot = tracer.getLatency();
+ assertEquals(snapshot.description(), "total: 2 ms, A: 999 us, B: 1 ms, C: 100 us");
+ assertEquals(snapshot.elapsedInMillis(), 2);
+ }
+
+ @Test
+ public void testTraceFuture() throws Exception {
+ LatencyTracer tracer = new LatencyTracer(new LinkedList<>(), System::nanoTime);
+ final CompletableFuture<Integer> future = CompletableFuture.completedFuture(100);
+ assertSame(tracer.trace("A", future), future);
+ final String latency = tracer.getLatency().description();
+ assertTrue(Pattern.compile("total: \\d+ ms").matcher(latency).matches(), latency);
+
+ final CompletableFuture<Integer> future2 = new CompletableFuture<Integer>();
+ delayedExecutor(500, TimeUnit.MILLISECONDS).execute(() -> future2.complete(1));
+ final CompletableFuture<Integer> tracedFuture = tracer.trace("B", future2);
+ assertNotSame(tracedFuture, future2);
+ assertEquals(tracedFuture.get(), 1);
+ final LatencyTracer.Snapshot snapshot = tracer.getLatency();
+ Matcher m = Pattern.compile("total: \\d+ ms, B: (\\d+) ms").matcher(snapshot.description());
+ assertTrue(m.matches(), snapshot.description());
+ assertEquals(Long.parseLong(m.group(1)), snapshot.elapsedInMillis(), snapshot.description());
+ assertTrue(snapshot.elapsedInMillis() >= 500, snapshot.description());
+ }
+
+ @Test
+ public void testTraceFailedFuture() throws Exception {
+ final LatencyTracer tracer = new LatencyTracer(new LinkedList<>(), testNanoTimeSupplier(1L));
+
+ final CompletableFuture<Void> future = new CompletableFuture<Void>();
+ delayedExecutor(100, TimeUnit.MILLISECONDS).execute(() -> future.completeExceptionally(
+ new RuntimeException("failure")));
+
+ try {
+ tracer.trace("A", future).get();
+ fail();
+ } catch (ExecutionException e) {
+ assertTrue(e.getCause() instanceof RuntimeException);
+ assertEquals(e.getCause().getMessage(), "failure");
+ }
+ }
+
+ // Replace it with `CompletableFuture.delayedExecutor` after upgraded to Java 9 or later
+ private static Executor delayedExecutor(long delay, TimeUnit unit) {
+
+ return command -> new Thread(() -> {
+ try {
+ Thread.sleep(unit.toMillis(delay));
+ command.run();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }).start();
+ }
+}