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 f618ed206b613..1ccd1478daf9e 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 @@ -1305,39 +1305,41 @@ public CompletableFuture> getTopic(final TopicName topicName, bo return FutureUtil.failedFuture(new NotAllowedException( "Broker is unable to load persistent topic")); } + final var timeoutSeconds = pulsar.getConfiguration().getTopicLoadTimeoutSeconds(); final CompletableFuture> 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() .attr("topic", topicName) - .attr("latencyMs", context.latencyMs(now)) - .log("Failed to load topic"); + .attr("latency", latency) + .log("Failed to load topic within " + timeoutSeconds + " s"); } else { log.warn() .attr("topic", topicName) - .attr("latencyMs", context.latencyString(now)) + .attr("latency", latency) .exception(t) .log("Failed to load topic"); } 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(); @@ -1349,7 +1351,7 @@ public CompletableFuture> getTopic(final TopicName topicName, bo // actual loading latency that should not be recorded in metrics. log.info() .attr("topic", topicName) - .attr("latencyMs", context.latencyString(System.nanoTime())) + .attr("latency", context.getLatency().description()) .log("Finished loading from other concurrent loading task"); cachedFuture.whenComplete((optTopic, e) -> { if (e == null) { @@ -1360,15 +1362,14 @@ public CompletableFuture> getTopic(final TopicName topicName, bo }); } }).exceptionally(e -> { - pulsar.getExecutor().execute(() -> topics.remove(topicName.toString(), topicFuture)); - final Throwable rc = FutureUtil.unwrapCompletionException(e); - log.error().attr("topic", topicName) - .exceptionMessage(rc) - .log("Topic creation encountered an exception" - + " by initialize topic policies service"); - topicFuture.completeExceptionally(rc); + log.warn().attr("topic", topicName).log("Topic creation encountered an exception" + + " by initialize topic policies service"); + failTopicFuture(topicName.toString(), topicFuture, e); return null; }); + }).exceptionally(e -> { + failTopicFuture(topicName.toString(), topicFuture, e); + return null; }); return topicFuture; } else { @@ -1420,6 +1421,14 @@ public CompletableFuture> getTopic(final TopicName topicName, bo } } + private void failTopicFuture(String topic, CompletableFuture> 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> getTopicPoliciesBypassSystemTopic(@NonNull TopicName topicName, TopicPoliciesService.GetType type) { if (ExtensibleLoadManagerImpl.isInternalTopic(topicName.toString())) { @@ -1859,7 +1868,7 @@ public PulsarAdmin getClusterPulsarAdmin(String cluster, Optional c protected CompletableFuture> loadOrCreatePersistentTopic(TopicLoadingContext context) { final var topic = context.getTopicName().toString(); final var topicFuture = context.getTopicFuture(); - checkTopicNsOwnership(topic) + context.trace("ownership", checkTopicNsOwnership(topic)) .thenRun(() -> { final Semaphore topicLoadSemaphore = topicLoadRequestSemaphore.get(); @@ -1871,8 +1880,7 @@ protected CompletableFuture> loadOrCreatePersistentTopic(TopicLo // 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(); @@ -1883,8 +1891,7 @@ protected CompletableFuture> loadOrCreatePersistentTopic(TopicLo log.debug().attr("topic", topic).log("topic-loading for added into pending queue"); } }).exceptionally(ex -> { - pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture)); - topicFuture.completeExceptionally(ex.getCause()); + failTopicFuture(topic, topicFuture, ex); return null; }); @@ -1924,7 +1931,7 @@ private void checkOwnershipAndCreatePersistentTopic(TopicLoadingContext context) 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> propertiesFuture; if (context.getProperties() == null) { //Read properties from storage when loading topic. @@ -1932,19 +1939,16 @@ private void checkOwnershipAndCreatePersistentTopic(TopicLoadingContext context) } 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().attr("topic", topic).exception(throwable).log("Read topic property failed"); - 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; }); } @@ -1957,10 +1961,8 @@ public void createPersistentTopic0(TopicLoadingContext context) { 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; } @@ -1977,6 +1979,7 @@ public void createPersistentTopic0(TopicLoadingContext context) { n.recycle(); return found; }), (managedLedgerConfig, exists) -> { + context.trace("ml-config"); if (isBrokerEntryMetadataEnabled() || isBrokerPayloadProcessorEnabled()) { // init managedLedger interceptor Set interceptors = new HashSet<>(); @@ -2038,27 +2041,26 @@ public void createPersistentTopic0(TopicLoadingContext context) { @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() .attr("topic", topic) .attr("dedupEnabled", persistentTopic.isDeduplicationEnabled()) - .attr("latencyMs", context.latencyString(nowInNanos)) + .attr("latency", latency.description()) .log("Loaded topic"); - pulsarStats.recordTopicLoadTimeValue(topic, topicLoadLatencyMs); + pulsarStats.recordTopicLoadTimeValue(topic, latency.elapsedInMillis()); if (!topicFuture.complete(Optional.of(persistentTopic))) { // Check create persistent topic timeout. if (topicFuture.isCompletedExceptionally()) { @@ -2110,12 +2112,7 @@ public void openLedgerComplete(ManagedLedger ledger, Object ctx) { return null; }); } catch (Exception e) { - log.warn() - .attr("topic", topic) - .exceptionMessage(e) - .log("Failed to create topic"); - pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture)); - topicFuture.completeExceptionally(e); + failTopicFuture(topic, topicFuture, e); } } @@ -2127,9 +2124,7 @@ public void openLedgerFailed(ManagedLedgerException exception, Object ctx) { loadFuture.completeExceptionally(exception); topicFuture.complete(Optional.empty()); } else { - log.warn().attr("topic", topic).exception(exception).log("Failed to create topic"); - pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture)); - topicFuture.completeExceptionally(new PersistenceException(exception)); + failTopicFuture(topic, topicFuture, new PersistenceException(exception)); } } }, () -> isTopicNsOwnedByBrokerAsync(topicName), null); @@ -2140,12 +2135,8 @@ public void openLedgerFailed(ManagedLedgerException exception, Object ctx) { "Failed to get topic configuration"; log.warn() .attr("topic", topic) - .exception(exception) .log(msg); - // 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); + failTopicFuture(topic, topicFuture, exception); return null; }); } @@ -3571,9 +3562,9 @@ private void createPendingLoadTopic() { return; } - pendingTopic.polledFromQueue(); + pendingTopic.trace("queued"); final String topic = pendingTopic.getTopicName().toString(); - checkTopicNsOwnership(topic).thenRun(() -> { + pendingTopic.trace("ownership", checkTopicNsOwnership(topic)).thenRun(() -> { CompletableFuture> 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 9e3ed230cd26b..b9a668f92b26c 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 @@ 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 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> 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 0000000000000..a1e23febb69a2 --- /dev/null +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/LatencyTracer.java @@ -0,0 +1,94 @@ +/* + * 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.Queue; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; + +public class LatencyTracer { + + private final Queue timepoints; + private final NanoTimeSupplier nanoTimeSupplier; + private final long startNs; + + public LatencyTracer(Queue timepoints, NanoTimeSupplier nanoTimeSupplier) { + this.timepoints = timepoints; + this.nanoTimeSupplier = nanoTimeSupplier; + this.startNs = nanoTimeSupplier.getNanos(); + } + + public CompletableFuture trace(String message, CompletableFuture 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 var timepoints = new ArrayList<>(this.timepoints); + if (timepoints.isEmpty()) { + return new Snapshot(startNs, 0, "total: 0 ms"); + } + final var sb = new StringBuilder(); + final var totalLatencyNs = TimeUnit.NANOSECONDS.toMillis(timepoints.get(timepoints.size() - 1).timeInNanos + - startNs); + sb.append("total: ").append(totalLatencyNs).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; + } + return new Snapshot(prevNs, totalLatencyNs, sb.toString()); + } + + public Snapshot traceAndGetLatency(String action) { + trace(action); + return getLatency(); + } + + public interface NanoTimeSupplier { + + long getNanos(); + } + + public record Timepoint(String name, long timeInNanos) { + } + + /** + * A snapshot of the latency tracer at a given moment. + * + * @param endTimeInNanos the latest traced timestamp in nanoseconds + * @param elapsedInMillis the elapsed time in milliseconds from when the tracer was created + * @param description the detailed description for traced latencies, e.g. "total: 100 ms, A: 60 ms, B: 40 ms" + */ + public record Snapshot(long endTimeInNanos, long elapsedInMillis, 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 0000000000000..af14265910fd7 --- /dev/null +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/LatencyTracerTest.java @@ -0,0 +1,129 @@ +/* + * 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.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 nanoTimesQueue = new LinkedList<>(); + + { + for (long nanoTime : nanoTimes) { + nanoTimesQueue.add(nanoTime); + } + } + + @Override + public long getNanos() { + final var nanos = nanoTimesQueue.poll(); + assertNotNull(nanos); + return nanos; + } + }; + } + + @Test + public void testMulti() { + final var tracer = new LatencyTracer(new LinkedList<>(), testNanoTimeSupplier(10_000_000L, 30_000_000L, + 70_000_000L, 80_000_000L)); + tracer.trace("A"); + tracer.trace("B"); + var 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() { + final var tracer = new LatencyTracer(new LinkedList<>(), testNanoTimeSupplier(0L, 20_000_000L)); + final var snapshot = tracer.getLatency(); + assertEquals(snapshot.description(), "total: 0 ms"); + assertEquals(snapshot.elapsedInMillis(), 0); + } + + @Test + public void testZeroMs() { + final var tracer = new LatencyTracer(new LinkedList<>(), testNanoTimeSupplier(0L, 999_999L, 2_000_000L, + 2_100_000L)); + tracer.trace("A"); + tracer.trace("B"); + tracer.trace("C"); + final var 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 { + final var tracer = new LatencyTracer(new LinkedList<>(), System::nanoTime); + final var future = CompletableFuture.completedFuture(100); + assertSame(tracer.trace("A", future), future); + final var latency = tracer.getLatency().description(); + assertTrue(Pattern.compile("total: \\d+ ms").matcher(latency).matches(), latency); + + final var future2 = new CompletableFuture(); + CompletableFuture.delayedExecutor(500, TimeUnit.MILLISECONDS).execute(() -> future2.complete(1)); + final var tracedFuture = tracer.trace("B", future2); + assertNotSame(tracedFuture, future2); + assertEquals(tracedFuture.get(), 1); + final var 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 var tracer = new LatencyTracer(new LinkedList<>(), testNanoTimeSupplier(1L)); + + final var future = new CompletableFuture(); + CompletableFuture.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"); + } + } +}