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/10/01 18:11:27 UTC
(pulsar) 37/38: [feat][broker] PIP-441:Add broker-level metrics for non-recoverable data skips (#24726)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch lh-branch-4.2-backports-2026-10-01
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d8d5f8c38e0f1cd9ad515bca2866f0f3c3421e88
Author: Penghui Li <pe...@apache.org>
AuthorDate: Thu Oct 1 03:02:43 2026 -0700
[feat][broker] PIP-441:Add broker-level metrics for non-recoverable data skips (#24726)
(cherry picked from commit f175a6c1804ac87dba1bee35638eabd07778c453)
---
.../bookkeeper/mledger/ManagedLedgerConfig.java | 3 +
.../mledger/NonRecoverableDataMetricsCallback.java | 42 ++++
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 14 +-
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 3 +
.../impl/NonRecoverableDataCallbackTest.java | 236 +++++++++++++++++++++
.../pulsar/broker/service/BrokerService.java | 17 ++
.../broker/stats/BrokerOperabilityMetrics.java | 59 ++++++
.../OpenTelemetryBrokerOperabilityStatsTest.java | 119 +++++++++++
8 files changed, 492 insertions(+), 1 deletion(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
index 907bc4712ecd..55d9223cbb81 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
@@ -127,6 +127,9 @@ public class ManagedLedgerConfig {
private String storageClassName;
@Getter
@Setter
+ private NonRecoverableDataMetricsCallback nonRecoverableDataMetricsCallback;
+ @Getter
+ @Setter
private String shadowSourceName;
@Getter
private boolean persistIndividualAckAsLongArray;
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/NonRecoverableDataMetricsCallback.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/NonRecoverableDataMetricsCallback.java
new file mode 100644
index 000000000000..21ea1ba8ef4c
--- /dev/null
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/NonRecoverableDataMetricsCallback.java
@@ -0,0 +1,42 @@
+/*
+ * 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.bookkeeper.mledger;
+
+import org.apache.bookkeeper.common.annotation.InterfaceAudience;
+import org.apache.bookkeeper.common.annotation.InterfaceStability;
+
+/**
+ * Callback interface for reporting metrics when non-recoverable data is skipped.
+ */
+@InterfaceAudience.LimitedPrivate
+@InterfaceStability.Stable
+public interface NonRecoverableDataMetricsCallback {
+
+ /**
+ * Called when a non-recoverable ledger is skipped.
+ * @param ledgerId the ledger ID that was skipped
+ */
+ void onSkipNonRecoverableLedger(long ledgerId);
+
+ /**
+ * Called when non-recoverable entries are skipped.
+ * @param entryCount the number of entries that were skipped
+ */
+ void onSkipNonRecoverableEntries(long entryCount);
+}
\ No newline at end of file
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
index 024b870bb6fc..8aaffeebf952 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -3199,8 +3199,16 @@ public class ManagedCursorImpl implements ManagedCursor {
log.warn("[{}] [{}] Since these entry for ledger [{}] is lost and the autoSkipNonRecoverableData is true, "
+ "these entries [{}:{}) will be auto acknowledge in subscription",
ledger.getName(), name, ledgerId, startEntryId, endEntryId);
+ AtomicLong skippedEntries = new AtomicLong();
asyncDelete(() -> LongStream.range(startEntryId, endEntryId)
- .mapToObj(i -> PositionFactory.create(ledgerId, i)).iterator(),
+ .mapToObj(i -> {
+ Position position = PositionFactory.create(ledgerId, i);
+ // asyncDelete consumes this iterator while holding the cursor write lock.
+ if (!internalIsMessageDeleted(position)) {
+ skippedEntries.incrementAndGet();
+ }
+ return position;
+ }).iterator(),
new AsyncCallbacks.DeleteCallback() {
@Override
public void deleteComplete(Object ctx) {
@@ -3214,6 +3222,10 @@ public class ManagedCursorImpl implements ManagedCursor {
// If the broker crashed, the non-recoverable ledger will be detected again.
}
}, null);
+ if (skippedEntries.get() > 0 && ledger.getConfig().getNonRecoverableDataMetricsCallback() != null) {
+ ledger.getConfig().getNonRecoverableDataMetricsCallback()
+ .onSkipNonRecoverableEntries(skippedEntries.get());
+ }
}
// //////////////////////////////////////////////////
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index 8089bbbe601f..c0a04e7188ae 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -2110,6 +2110,9 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
for (ManagedCursor managedCursor : cursors) {
managedCursor.skipNonRecoverableLedger(ledgerId);
}
+ if (config.getNonRecoverableDataMetricsCallback() != null) {
+ config.getNonRecoverableDataMetricsCallback().onSkipNonRecoverableLedger(ledgerId);
+ }
}
synchronized void createLedgerAfterClosed() {
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/NonRecoverableDataCallbackTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/NonRecoverableDataCallbackTest.java
new file mode 100644
index 000000000000..bead9137464c
--- /dev/null
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/NonRecoverableDataCallbackTest.java
@@ -0,0 +1,236 @@
+/*
+ * 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.bookkeeper.mledger.impl;
+
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.testng.Assert.assertEquals;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+import lombok.Cleanup;
+import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
+import org.apache.bookkeeper.mledger.NonRecoverableDataMetricsCallback;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.PositionFactory;
+import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
+import org.testng.annotations.Test;
+
+/**
+ * Test the NonRecoverableDataMetricsCallback integration in ManagedLedgerImpl and ManagedCursorImpl.
+ */
+public class NonRecoverableDataCallbackTest extends MockedBookKeeperTestCase {
+
+ @Test
+ public void testManagedLedgerSkipNonRecoverableLedgerCallback() throws Exception {
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+
+ // Create a mock callback to track invocations
+ NonRecoverableDataMetricsCallback mockCallback = mock(NonRecoverableDataMetricsCallback.class);
+ config.setNonRecoverableDataMetricsCallback(mockCallback);
+
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("test-ledger", config);
+
+ // Call skipNonRecoverableLedger - this should trigger the callback
+ long ledgerId = 12345L;
+ ledger.skipNonRecoverableLedger(ledgerId);
+
+ // Verify the callback was called with the correct ledger ID
+ verify(mockCallback, times(1)).onSkipNonRecoverableLedger(eq(ledgerId));
+ verify(mockCallback, never()).onSkipNonRecoverableEntries(anyLong());
+
+ ledger.close();
+ }
+
+ @Test
+ public void testManagedLedgerSkipNonRecoverableLedgerWithoutCallback() throws Exception {
+ // Test that skipNonRecoverableLedger works when no callback is set
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ // Don't set callback - should be null by default
+
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("test-ledger-no-callback", config);
+
+ // This should not throw an exception even with null callback
+ ledger.skipNonRecoverableLedger(12345L);
+
+ ledger.close();
+ }
+
+ @Test
+ public void testManagedCursorSkipNonRecoverableEntriesCallback() throws Exception {
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+
+ // Create a callback that counts the entries
+ AtomicLong entriesSkipped = new AtomicLong(0);
+ config.setNonRecoverableDataMetricsCallback(new NonRecoverableDataMetricsCallback() {
+ @Override
+ public void onSkipNonRecoverableLedger(long ledgerId) {
+ // Not used in this test
+ }
+
+ @Override
+ public void onSkipNonRecoverableEntries(long entryCount) {
+ entriesSkipped.addAndGet(entryCount);
+ }
+ });
+
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("test-cursor-ledger", config);
+
+ // Test the callback directly by accessing it through the config
+ // This verifies that the callback is properly set and can be invoked
+ if (ledger.getConfig().getNonRecoverableDataMetricsCallback() != null) {
+ ledger.getConfig().getNonRecoverableDataMetricsCallback().onSkipNonRecoverableEntries(5L);
+ }
+
+ // Verify the callback was called with the expected count
+ assertEquals(entriesSkipped.get(), 5L);
+
+ ledger.close();
+ }
+
+ @Test
+ public void testBatchedEntrySkipCountsOnlyUnacknowledgedEntries() throws Exception {
+ NonRecoverableDataMetricsCallback callback = mock(NonRecoverableDataMetricsCallback.class);
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setNonRecoverableDataMetricsCallback(callback);
+ @Cleanup
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("test-batched-entry-skips", config);
+ @Cleanup
+ ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("cursor");
+ List<Position> positions = new ArrayList<>();
+ for (int i = 0; i < 5; i++) {
+ positions.add(ledger.addEntry(new byte[] {1}));
+ }
+ cursor.delete(positions.get(2));
+ Position end = PositionFactory.create(positions.get(4).getLedgerId(), positions.get(4).getEntryId() + 1);
+
+ cursor.skipNonRecoverableEntries(positions.get(1), end);
+ verify(callback).onSkipNonRecoverableEntries(3L);
+ cursor.skipNonRecoverableEntries(positions.get(1), end);
+ verify(callback, times(1)).onSkipNonRecoverableEntries(anyLong());
+
+ cursor.skipNonRecoverableEntries(positions.get(0), end);
+ verify(callback).onSkipNonRecoverableEntries(1L);
+ cursor.skipNonRecoverableEntries(positions.get(0), end);
+ verify(callback, times(2)).onSkipNonRecoverableEntries(anyLong());
+ }
+
+ @Test
+ public void testManagedCursorSkipNonRecoverableEntriesWithoutCallback() throws Exception {
+ // Test that the method works when no callback is set
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ // Don't set callback - should be null by default
+
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("test-cursor-no-callback", config);
+
+ // Verify the callback is null
+ assertEquals(ledger.getConfig().getNonRecoverableDataMetricsCallback(), null);
+
+ ledger.close();
+ }
+
+ @Test
+ public void testMultipleLedgerSkips() throws Exception {
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+
+ NonRecoverableDataMetricsCallback mockCallback = mock(NonRecoverableDataMetricsCallback.class);
+ config.setNonRecoverableDataMetricsCallback(mockCallback);
+
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("test-multiple-skips", config);
+
+ // Skip multiple ledgers
+ ledger.skipNonRecoverableLedger(100L);
+ ledger.skipNonRecoverableLedger(200L);
+ ledger.skipNonRecoverableLedger(300L);
+
+ // Verify callback was called for each ledger
+ verify(mockCallback, times(1)).onSkipNonRecoverableLedger(eq(100L));
+ verify(mockCallback, times(1)).onSkipNonRecoverableLedger(eq(200L));
+ verify(mockCallback, times(1)).onSkipNonRecoverableLedger(eq(300L));
+ verify(mockCallback, never()).onSkipNonRecoverableEntries(anyLong());
+
+ ledger.close();
+ }
+
+ @Test
+ public void testMultipleEntrySkips() throws Exception {
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+
+ AtomicLong totalEntriesSkipped = new AtomicLong(0);
+ AtomicLong callbackInvocations = new AtomicLong(0);
+
+ config.setNonRecoverableDataMetricsCallback(new NonRecoverableDataMetricsCallback() {
+ @Override
+ public void onSkipNonRecoverableLedger(long ledgerId) {
+ // Not used in this test
+ }
+
+ @Override
+ public void onSkipNonRecoverableEntries(long entryCount) {
+ totalEntriesSkipped.addAndGet(entryCount);
+ callbackInvocations.incrementAndGet();
+ }
+ });
+
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("test-multiple-entry-skips", config);
+
+ // Test multiple invocations of the callback directly
+ var callback = ledger.getConfig().getNonRecoverableDataMetricsCallback();
+ if (callback != null) {
+ callback.onSkipNonRecoverableEntries(5L); // 5 entries
+ callback.onSkipNonRecoverableEntries(2L); // 2 entries
+ callback.onSkipNonRecoverableEntries(5L); // 5 entries
+ }
+
+ // Verify callback was invoked multiple times and total entries are correct
+ assertEquals(callbackInvocations.get(), 3);
+ assertEquals(totalEntriesSkipped.get(), 12L); // 5 + 2 + 5 = 12
+
+ ledger.close();
+ }
+
+ @Test
+ public void testSkipZeroEntries() throws Exception {
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+
+ NonRecoverableDataMetricsCallback mockCallback = mock(NonRecoverableDataMetricsCallback.class);
+ config.setNonRecoverableDataMetricsCallback(mockCallback);
+
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("test-zero-entries", config);
+
+ // Verify that the callback is not invoked when 0 entries are skipped
+ // In real scenarios, the skipNonRecoverableEntries method checks the count and only
+ // calls the callback if skippedEntries > 0
+ var callback = ledger.getConfig().getNonRecoverableDataMetricsCallback();
+ if (callback != null) {
+ // This simulates the condition where no entries are actually skipped
+ // The method would not call the callback in this case
+ }
+
+ // The callback should not be invoked for zero entry counts (we don't call it)
+ verify(mockCallback, never()).onSkipNonRecoverableEntries(anyLong());
+ verify(mockCallback, never()).onSkipNonRecoverableLedger(anyLong());
+
+ ledger.close();
+ }
+}
\ No newline at end of file
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 c666f112a027..f7ba8536b823 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
@@ -1927,6 +1927,23 @@ public class BrokerService implements Closeable {
managedLedgerConfig.setManagedLedgerInterceptor(
new ManagedLedgerInterceptorImpl(interceptors, brokerEntryPayloadProcessors));
}
+
+ // Set non-recoverable data metrics callback
+ if (pulsarStats.getBrokerOperabilityMetrics() != null) {
+ managedLedgerConfig.setNonRecoverableDataMetricsCallback(
+ new org.apache.bookkeeper.mledger.NonRecoverableDataMetricsCallback() {
+ @Override
+ public void onSkipNonRecoverableLedger(long ledgerId) {
+ pulsarStats.getBrokerOperabilityMetrics().recordNonRecoverableLedgerSkipped();
+ }
+
+ @Override
+ public void onSkipNonRecoverableEntries(long entryCount) {
+ pulsarStats.getBrokerOperabilityMetrics().recordNonRecoverableEntriesSkipped(entryCount);
+ }
+ });
+ }
+
managedLedgerConfig.setCreateIfMissing(createIfMissing);
if (context.getProperties() != null) {
managedLedgerConfig.setProperties(context.getProperties());
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/BrokerOperabilityMetrics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/BrokerOperabilityMetrics.java
index 310c14d4afa3..33aca625e66b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/BrokerOperabilityMetrics.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/BrokerOperabilityMetrics.java
@@ -39,6 +39,12 @@ import org.apache.pulsar.opentelemetry.annotations.PulsarDeprecatedMetric;
*/
public class BrokerOperabilityMetrics implements AutoCloseable {
private static final Counter TOPIC_LOAD_FAILED = Counter.build("topic_load_failed", "-").register();
+ private static final Counter NON_RECOVERABLE_LEDGERS_SKIPPED = Counter.build(
+ "pulsar_broker_non_recoverable_ledgers_skipped_total",
+ "Count of non-recoverable ledgers skipped").register();
+ private static final Counter NON_RECOVERABLE_ENTRIES_SKIPPED = Counter.build(
+ "pulsar_broker_non_recoverable_entries_skipped_total",
+ "Count of non-recoverable entries skipped").register();
private final List<Metrics> metricsList;
private final String localCluster;
private final DimensionStats topicLoadStats;
@@ -50,6 +56,8 @@ public class BrokerOperabilityMetrics implements AutoCloseable {
private final LongAdder connectionCreateSuccessCount;
private final LongAdder connectionCreateFailCount;
+ private final LongAdder nonRecoverableLedgersSkippedCount;
+ private final LongAdder nonRecoverableEntriesSkippedCount;
public static final String CONNECTION_COUNTER_METRIC_NAME = "pulsar.broker.connection.count";
private final ObservableLongCounter connectionCounter;
@@ -58,6 +66,14 @@ public class BrokerOperabilityMetrics implements AutoCloseable {
"pulsar.broker.connection.create.operation.count";
private final ObservableLongCounter connectionCreateCounter;
+ public static final String NON_RECOVERABLE_LEDGERS_SKIPPED_COUNTER_METRIC_NAME =
+ "pulsar.broker.non.recoverable.ledgers.skipped.count";
+ private final ObservableLongCounter nonRecoverableLedgersSkippedCounter;
+
+ public static final String NON_RECOVERABLE_ENTRIES_SKIPPED_COUNTER_METRIC_NAME =
+ "pulsar.broker.non.recoverable.entries.skipped.count";
+ private final ObservableLongCounter nonRecoverableEntriesSkippedCounter;
+
public static final String TOPIC_PUBLISH_LATENCY_METRIC_NAME = "pulsar.broker.topic.publish.latency";
private final DoubleHistogram topicPublishLatencyHistogram;
@PulsarDeprecatedMetric(newMetricName = TOPIC_PUBLISH_LATENCY_METRIC_NAME)
@@ -82,6 +98,8 @@ public class BrokerOperabilityMetrics implements AutoCloseable {
this.healthCheckStatus = -1;
this.connectionCreateSuccessCount = new LongAdder();
this.connectionCreateFailCount = new LongAdder();
+ this.nonRecoverableLedgersSkippedCount = new LongAdder();
+ this.nonRecoverableEntriesSkippedCount = new LongAdder();
connectionCounter = pulsar.getOpenTelemetry().getMeter()
.counterBuilder(CONNECTION_COUNTER_METRIC_NAME)
@@ -105,6 +123,22 @@ public class BrokerOperabilityMetrics implements AutoCloseable {
measurement.record(connectionCreateFailCount.sum(), ConnectionCreateStatus.FAILURE.attributes);
});
+ nonRecoverableLedgersSkippedCounter = pulsar.getOpenTelemetry().getMeter()
+ .counterBuilder(NON_RECOVERABLE_LEDGERS_SKIPPED_COUNTER_METRIC_NAME)
+ .setDescription("The number of non-recoverable ledgers skipped.")
+ .setUnit("{ledger}")
+ .buildWithCallback(measurement -> {
+ measurement.record(nonRecoverableLedgersSkippedCount.sum());
+ });
+
+ nonRecoverableEntriesSkippedCounter = pulsar.getOpenTelemetry().getMeter()
+ .counterBuilder(NON_RECOVERABLE_ENTRIES_SKIPPED_COUNTER_METRIC_NAME)
+ .setDescription("The number of non-recoverable entries skipped.")
+ .setUnit("{entry}")
+ .buildWithCallback(measurement -> {
+ measurement.record(nonRecoverableEntriesSkippedCount.sum());
+ });
+
this.topicPublishLatencyHistogram = pulsar.getOpenTelemetry().getMeter()
.histogramBuilder(TOPIC_PUBLISH_LATENCY_METRIC_NAME)
.setUnit("s")
@@ -118,6 +152,8 @@ public class BrokerOperabilityMetrics implements AutoCloseable {
public void close() throws Exception {
connectionCounter.close();
connectionCreateCounter.close();
+ nonRecoverableLedgersSkippedCounter.close();
+ nonRecoverableEntriesSkippedCounter.close();
}
public List<Metrics> getMetrics() {
@@ -130,6 +166,7 @@ public class BrokerOperabilityMetrics implements AutoCloseable {
metricsList.add(getTopicLoadMetrics());
metricsList.add(getConnectionMetrics());
metricsList.add(getHealthMetrics());
+ metricsList.add(getNonRecoverableSkippedMetrics());
}
public Metrics generateConnectionMetrics() {
@@ -152,6 +189,13 @@ public class BrokerOperabilityMetrics implements AutoCloseable {
return rMetrics;
}
+ Metrics getNonRecoverableSkippedMetrics() {
+ Metrics rMetrics = Metrics.create(getDimensionMap("broker_non_recoverable_skipped"));
+ rMetrics.put("brk_non_recoverable_ledgers_skipped_total", NON_RECOVERABLE_LEDGERS_SKIPPED.get());
+ rMetrics.put("brk_non_recoverable_entries_skipped_total", NON_RECOVERABLE_ENTRIES_SKIPPED.get());
+ return rMetrics;
+ }
+
Map<String, String> getDimensionMap(String metricsName) {
Map<String, String> dimensionMap = new HashMap<>();
dimensionMap.put("broker", brokerName);
@@ -221,6 +265,21 @@ public class BrokerOperabilityMetrics implements AutoCloseable {
this.healthCheckStatus = 0;
}
+ public void recordNonRecoverableLedgerSkipped() {
+ this.nonRecoverableLedgersSkippedCount.increment();
+ NON_RECOVERABLE_LEDGERS_SKIPPED.inc();
+ }
+
+ public void recordNonRecoverableEntriesSkipped() {
+ this.nonRecoverableEntriesSkippedCount.increment();
+ NON_RECOVERABLE_ENTRIES_SKIPPED.inc();
+ }
+
+ public void recordNonRecoverableEntriesSkipped(long amount) {
+ this.nonRecoverableEntriesSkippedCount.add(amount);
+ NON_RECOVERABLE_ENTRIES_SKIPPED.inc(amount);
+ }
+
public void recordPublishLatency(long latency, TimeUnit unit) {
this.topicPublishLatencyHistogram.record(unit.toMillis(latency) / 1000.0);
PUBLISH_LATENCY.observe(latency, unit);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/OpenTelemetryBrokerOperabilityStatsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/OpenTelemetryBrokerOperabilityStatsTest.java
index 4991c0e5719e..fadfb77832df 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/OpenTelemetryBrokerOperabilityStatsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/OpenTelemetryBrokerOperabilityStatsTest.java
@@ -23,11 +23,24 @@ import static org.apache.pulsar.broker.stats.BrokerOpenTelemetryTestUtil.assertM
import static org.apache.pulsar.broker.stats.BrokerOpenTelemetryTestUtil.assertMetricLongSumValue;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.sdk.metrics.data.MetricData;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
import java.util.concurrent.TimeUnit;
import lombok.Cleanup;
+import org.apache.bookkeeper.mledger.ManagedLedger;
+import org.apache.bookkeeper.mledger.impl.ImmutablePositionImpl;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.service.BrokerTestBase;
+import org.apache.pulsar.broker.service.Topic;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.testcontext.PulsarTestContext;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.MessageIdAdv;
+import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes;
@@ -115,6 +128,112 @@ public class OpenTelemetryBrokerOperabilityStatsTest extends BrokerTestBase {
ConnectionCreateStatus.FAILURE.attributes, 1);
}
+ @Test
+ public void testNonRecoverableDataMetrics() throws Exception {
+ BrokerOperabilityMetrics brokerOperabilityMetrics = pulsar.getBrokerService()
+ .getPulsarStats().getBrokerOperabilityMetrics();
+
+ // Test initial state - should be 0
+ var metrics = pulsarTestContext.getOpenTelemetryMetricReader().collectAllMetrics();
+ assertMetricLongSumValue(metrics, BrokerOperabilityMetrics.NON_RECOVERABLE_LEDGERS_SKIPPED_COUNTER_METRIC_NAME,
+ Attributes.empty(), 0);
+ assertMetricLongSumValue(metrics, BrokerOperabilityMetrics.NON_RECOVERABLE_ENTRIES_SKIPPED_COUNTER_METRIC_NAME,
+ Attributes.empty(), 0);
+
+ // Test recording ledger skip metrics
+ brokerOperabilityMetrics.recordNonRecoverableLedgerSkipped();
+ brokerOperabilityMetrics.recordNonRecoverableLedgerSkipped();
+
+ // Test recording entry skip metrics
+ brokerOperabilityMetrics.recordNonRecoverableEntriesSkipped();
+ brokerOperabilityMetrics.recordNonRecoverableEntriesSkipped();
+ brokerOperabilityMetrics.recordNonRecoverableEntriesSkipped();
+
+ // Verify the metrics have been updated
+ metrics = pulsarTestContext.getOpenTelemetryMetricReader().collectAllMetrics();
+ assertMetricLongSumValue(metrics, BrokerOperabilityMetrics.NON_RECOVERABLE_LEDGERS_SKIPPED_COUNTER_METRIC_NAME,
+ Attributes.empty(), 2);
+ assertMetricLongSumValue(metrics, BrokerOperabilityMetrics.NON_RECOVERABLE_ENTRIES_SKIPPED_COUNTER_METRIC_NAME,
+ Attributes.empty(), 3);
+ }
+
+ @Test
+ public void testEndToEndManagedLedgerCallbackIntegration() throws Exception {
+ BrokerOperabilityMetrics brokerOperabilityMetrics = pulsar.getBrokerService()
+ .getPulsarStats().getBrokerOperabilityMetrics();
+
+ // Get initial metric values
+ Collection<MetricData> initialMetrics = pulsarTestContext.getOpenTelemetryMetricReader().collectAllMetrics();
+ long initialLedgersSkipped = getMetricValue(initialMetrics,
+ BrokerOperabilityMetrics.NON_RECOVERABLE_LEDGERS_SKIPPED_COUNTER_METRIC_NAME);
+ long initialEntriesSkipped = getMetricValue(initialMetrics,
+ BrokerOperabilityMetrics.NON_RECOVERABLE_ENTRIES_SKIPPED_COUNTER_METRIC_NAME);
+
+ // Test end-to-end callback integration by creating a real topic and subscription
+ // This ensures we're testing the actual callback that BrokerService sets up during topic creation
+ String topicName = BrokerTestUtil.newUniqueName("persistent://my-property/my-ns/testEndToEndCallback");
+
+ // Create a consumer, which will automatically create the topic and subscription
+ @Cleanup
+ Consumer<byte[]> consumer = pulsarClient.newConsumer()
+ .topic(topicName)
+ .subscriptionName("test-subscription")
+ .subscribe();
+ @Cleanup
+ Producer<byte[]> producer = pulsarClient.newProducer()
+ .topic(topicName)
+ .enableBatching(false)
+ .create();
+
+ List<MessageIdAdv> messageIds = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ messageIds.add((MessageIdAdv) producer.newMessage().value(("message-" + i).getBytes()).send());
+ }
+
+ // Get the topic instance from the broker service
+ Topic topic = pulsar.getBrokerService().getTopicIfExists(topicName).get().get();
+ PersistentTopic persistentTopic = (PersistentTopic) topic;
+
+ // Get the managed ledger and cursor from the topic - these have the callback configured by BrokerService
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ PersistentSubscription subscription = persistentTopic.getSubscription("test-subscription");
+ ManagedCursorImpl cursor = (ManagedCursorImpl) subscription.getCursor();
+
+ long ledgerId = messageIds.get(0).getLedgerId();
+ long firstEntryId = messageIds.get(0).getEntryId();
+ long lastEntryId = messageIds.get(messageIds.size() - 1).getEntryId();
+
+ // Test skipNonRecoverableLedger - this should trigger the callback set up by BrokerService
+ managedLedger.skipNonRecoverableLedger(ledgerId);
+ cursor.skipNonRecoverableEntries(new ImmutablePositionImpl(ledgerId, firstEntryId),
+ new ImmutablePositionImpl(ledgerId, lastEntryId + 1));
+
+ // Verify the metrics were updated through the callback chain
+ Collection<MetricData> finalMetrics = pulsarTestContext.getOpenTelemetryMetricReader().collectAllMetrics();
+
+ // Verify that the ledger skip metrics increased by 1 (we called skipNonRecoverableLedger once)
+ assertMetricLongSumValue(finalMetrics,
+ BrokerOperabilityMetrics.NON_RECOVERABLE_LEDGERS_SKIPPED_COUNTER_METRIC_NAME,
+ Attributes.empty(), initialLedgersSkipped + 1);
+
+ // Verify that the entries skip metrics increased by 10 (10 entries skipped)
+ assertMetricLongSumValue(finalMetrics,
+ BrokerOperabilityMetrics.NON_RECOVERABLE_ENTRIES_SKIPPED_COUNTER_METRIC_NAME,
+ Attributes.empty(), initialEntriesSkipped + 10);
+ }
+
+ private long getMetricValue(Collection<MetricData> metrics,
+ String metricName) {
+ return metrics.stream()
+ .filter(m -> m.getName().equals(metricName))
+ .findFirst()
+ .map(metricData -> metricData.getLongSumData().getPoints().stream()
+ .findFirst()
+ .map(point -> point.getValue())
+ .orElse(0L))
+ .orElse(0L);
+ }
+
@Test
public void testPublishLatency() throws Exception {
final var topicName = BrokerTestUtil.newUniqueName("persistent://my-property/my-ns/testPublishLatency");