Denovo1998 commented on code in PR #26163: URL: https://github.com/apache/pulsar/pull/26163#discussion_r3570251815
########## pulsar-common/src/main/java/org/apache/pulsar/common/util/LatencyTracer.java: ########## @@ -0,0 +1,101 @@ +/* + * 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.Queue; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLongFieldUpdater; + +public class LatencyTracer { + + private static final AtomicLongFieldUpdater<LatencyTracer> END_TIME_UPDATER = AtomicLongFieldUpdater.newUpdater( + LatencyTracer.class, "endNs"); + private final Queue<Timepoint> timepoints; + private final NanoTimeSupplier nanoTimeSupplier; + private final long startNs; + private volatile long endNs = -1L; + + 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) { Review Comment: Calling `latencyString()` freezes `endNs`, but `trace()` still accepts checkpoints afterwards. This happens in the current BrokerService flow: a timeout logs the latency without cancelling the underlying topic load, and the remaining stages can continue tracing and later reach the success callback. A deterministic test that snapshots at 10 ms and traces B at 20 ms produces `total: 10 ms, A: 1 ms, B: 19 ms, done: -10000 us`. Could `latencyString()` create a local immutable snapshot instead of permanently ending the tracer? Alternatively, if the tracer has explicit terminal semantics, `trace()` must reject/ignore checkpoints after termination, but that would lose the post-timeout loading diagnostics. ########## pulsar-common/src/main/java/org/apache/pulsar/common/util/LatencyTracer.java: ########## @@ -0,0 +1,101 @@ +/* + * 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.Queue; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLongFieldUpdater; + +public class LatencyTracer { + + private static final AtomicLongFieldUpdater<LatencyTracer> END_TIME_UPDATER = AtomicLongFieldUpdater.newUpdater( + LatencyTracer.class, "endNs"); + private final Queue<Timepoint> timepoints; + private final NanoTimeSupplier nanoTimeSupplier; + private final long startNs; + private volatile long endNs = -1L; + + 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 long latencyInMillis() { + ensureEndTimeSet(); + return TimeUnit.NANOSECONDS.toMillis(endNs >= 0L ? endNs - startNs : System.nanoTime() - startNs); + } + + public String latencyString() { + ensureEndTimeSet(); + StringBuilder sb = new StringBuilder(); + sb.append("total: ").append(latencyInMillis()).append(" ms"); + long prevNs = startNs; + for (final var 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; + } + if (prevNs != startNs) { + sb.append(", done: "); + final var latencyMs = TimeUnit.NANOSECONDS.toMillis(endNs - prevNs); + if (latencyMs > 0) { + sb.append(latencyMs).append(" ms"); + } else { + sb.append(TimeUnit.NANOSECONDS.toMicros(endNs - prevNs)).append(" us"); + } + } + return sb.toString(); + } + + private void ensureEndTimeSet() { Review Comment: The `-2` initialization state is observable by other callers. If one thread wins the CAS and blocks inside `nanoTimeSupplier.getNanos()`, another thread returns from `ensureEndTimeSet()` with `endNs == -2`. `latencyInMillis()` then mixes in `System.nanoTime()`, while `latencyString()` uses `-2` directly for the `done` delta. Could we avoid publishing an intermediate timestamp state? A local immutable snapshot would also avoid making readers wait for or observe partially initialized state. ########## pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java: ########## @@ -1313,31 +1313,31 @@ public CompletableFuture<Optional<Topic>> getTopic(final TopicName topicName, bo context.setProperties(properties); } topicFuture.exceptionally(t -> { - final var now = System.nanoTime(); if (FutureUtil.unwrapCompletionException(t) instanceof TimeoutException) { log.warn() .attr("topic", topicName) - .attr("latencyMs", context.latencyMs(now)) + .attr("latency", context.latencyString()) .log("Failed to load topic"); } else { log.warn() .attr("topic", topicName) - .attr("latencyMs", context.latencyString(now)) + .attr("latency", context.latencyString()) .exception(t) .log("Failed to load topic"); } pulsarStats.recordTopicLoadFailed(); return Optional.empty(); }); - checkNonPartitionedTopicExists(topicName).thenAccept(exists -> { + context.trace("topic exists", checkNonPartitionedTopicExists(topicName)).thenAccept(exists -> { Review Comment: If `checkNonPartitionedTopicExists()` completes exceptionally, `thenAccept` is skipped and the exception is not propagated to `topicFuture`. The caller therefore waits for the topic-load timeout and receives a timeout instead of the original metadata-store failure. Could we attach an exception handler to this chain and complete `topicFuture` with the unwrapped cause? ########## pulsar-common/src/main/java/org/apache/pulsar/common/util/LatencyTracer.java: ########## @@ -0,0 +1,101 @@ +/* + * 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.Queue; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLongFieldUpdater; + +public class LatencyTracer { + + private static final AtomicLongFieldUpdater<LatencyTracer> END_TIME_UPDATER = AtomicLongFieldUpdater.newUpdater( + LatencyTracer.class, "endNs"); + private final Queue<Timepoint> timepoints; + private final NanoTimeSupplier nanoTimeSupplier; + private final long startNs; + private volatile long endNs = -1L; + + 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 long latencyInMillis() { + ensureEndTimeSet(); + return TimeUnit.NANOSECONDS.toMillis(endNs >= 0L ? endNs - startNs : System.nanoTime() - startNs); + } + + public String latencyString() { + ensureEndTimeSet(); + StringBuilder sb = new StringBuilder(); + sb.append("total: ").append(latencyInMillis()).append(" ms"); + long prevNs = startNs; + for (final var 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; + } + if (prevNs != startNs) { + sb.append(", done: "); + final var latencyMs = TimeUnit.NANOSECONDS.toMillis(endNs - prevNs); + if (latencyMs > 0) { + sb.append(latencyMs).append(" ms"); + } else { + sb.append(TimeUnit.NANOSECONDS.toMicros(endNs - prevNs)).append(" us"); + } + } + return sb.toString(); + } + + private void ensureEndTimeSet() { Review Comment: There is also an issue here: `System.nanoTime()` is explicitly allowed to return negative values, so `endNs >= 0` cannot distinguish a valid timestamp from an unset/in-progress state, and `-1`/`-2` can collide with valid clock values. With an injected start of -20 ms and end of -10 ms, `latencyString()` should report 10 ms but instead falls back to the unrelated `System.nanoTime()` clock and reports a huge value. State should be represented separately from the timestamp. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
