Added unit tests

pulsar-broker/pom.xml

@@ -149,6 +149,11 @@
       <version>${project.version}</version>
     </dependency>
 
+    <dependency>
+      <groupId>io.opentelemetry</groupId>
+      <artifactId>opentelemetry-sdk-testing</artifactId>
+    </dependency>
+
     <dependency>
       <groupId>${project.groupId}</groupId>
       <artifactId>pulsar-io-batch-discovery-triggerers</artifactId>

pulsar-broker/src/test/java/org/apache/pulsar/client/metrics/ClientMetricsTest.java

@@ -0,0 +1,333 @@
+/*
+ * 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.client.metrics;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.metrics.SdkMeterProvider;
+import io.opentelemetry.sdk.metrics.data.MetricData;
+import io.opentelemetry.sdk.metrics.data.MetricDataType;
+import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.concurrent.TimeUnit;
+import lombok.Cleanup;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker-api")
+public class ClientMetricsTest extends ProducerConsumerBase {
+
+    InMemoryMetricReader reader;
+    OpenTelemetry otel;
+
+    @BeforeMethod
+    @Override
+    protected void setup() throws Exception {
+        super.internalSetup();
+        super.producerBaseSetup();
+
+        this.reader = InMemoryMetricReader.create();
+        SdkMeterProvider sdkMeterProvider = SdkMeterProvider.builder()
+                .registerMetricReader(reader)
+                .build();
+        this.otel = OpenTelemetrySdk.builder().setMeterProvider(sdkMeterProvider).build();
+    }
+
+    @AfterMethod(alwaysRun = true)
+    @Override
+    protected void cleanup() throws Exception {
+        super.internalCleanup();
+    }
+
+    private Map<String, MetricData> collectMetrics() {
+        Map<String, MetricData> metrics = new TreeMap<>();
+        for (MetricData md : reader.collectAllMetrics()) {
+            metrics.put(md.getName(), md);
+        }
+        return metrics;
+    }
+
+    private void assertCounterValue(Map<String, MetricData> metrics, String name, long expectedValue,
+                                    Attributes expectedAttributes) {
+        assertEquals(getCounterValue(metrics, name, expectedAttributes), expectedValue);
+    }
+
+    private long getCounterValue(Map<String, MetricData> metrics, String name,
+                                    Attributes expectedAttributes) {
+        MetricData md = metrics.get(name);
+        assertNotNull(md, "metric not found: " + name);
+        assertEquals(md.getType(), MetricDataType.LONG_SUM);
+
+        for (var ex : md.getLongSumData().getPoints()) {
+            if (ex.getAttributes().equals(expectedAttributes)) {
+                return ex.getValue();
+            }
+        }
+
+        fail("metric attributes not found: " + expectedAttributes);
+        return -1;
+    }
+
+    private void assertHistoCountValue(Map<String, MetricData> metrics, String name, long expectedCount,
+                                       Attributes expectedAttributes) {
+        assertEquals(getHistoCountValue(metrics, name, expectedAttributes), expectedCount);
+    }
+
+    private long getHistoCountValue(Map<String, MetricData> metrics, String name,
+                                    Attributes expectedAttributes) {
+        MetricData md = metrics.get(name);
+        assertNotNull(md, "metric not found: " + name);
+        assertEquals(md.getType(), MetricDataType.HISTOGRAM);
+
+        for (var ex : md.getHistogramData().getPoints()) {
+            if (ex.getAttributes().equals(expectedAttributes)) {
+                return ex.getCount();
+            }
+        }
+
+        fail("metric attributes not found: " + expectedAttributes);
+        return -1;
+    }
+
+    @Test
+    public void testProducerMetrics() throws Exception {
+        String topic = newTopicName();
+
+        PulsarClient client = PulsarClient.builder()
+                .serviceUrl(pulsar.getBrokerServiceUrl())
+                .openTelemetry(otel)
+                .build();
+
+        Producer<String> producer = client.newProducer(Schema.STRING)
+                .topic(topic)
+                .create();
+
+        for (int i = 0; i < 5; i++) {
+            producer.send("Hello");
+        }
+
+        Attributes nsAttrs = Attributes.builder()
+                .put("pulsar.tenant", "my-property")
+                .put("pulsar.namespace", "my-property/my-ns")
+                .build();
+        Attributes nsAttrsSuccess = nsAttrs.toBuilder()
+                .put("success", true)
+                .build();
+
+        var metrics = collectMetrics();
+        System.err.println("All metrics: " + metrics.keySet());
+
+        assertCounterValue(metrics, "pulsar.client.connections.opened", 1, Attributes.empty());
+        assertCounterValue(metrics, "pulsar.client.producer.message.pending.count", 0, nsAttrs);
+        assertCounterValue(metrics, "pulsar.client.producer.message.pending.size", 0, nsAttrs);
+
+        assertHistoCountValue(metrics, "pulsar.client.lookup.duration", 1,
+                Attributes.builder()
+                        .put("pulsar.lookup.transport-type", "binary")
+                        .put("pulsar.lookup.type", "topic")
+                        .put("success", true)
+                        .build());
+        assertHistoCountValue(metrics, "pulsar.client.lookup.duration", 1,
+                Attributes.builder()
+                        .put("pulsar.lookup.transport-type", "binary")
+                        .put("pulsar.lookup.type", "metadata")
+                        .put("success", true)
+                        .build());
+
+        assertHistoCountValue(metrics, "pulsar.client.producer.message.send.duration", 5, nsAttrsSuccess);
+        assertHistoCountValue(metrics, "pulsar.client.producer.rpc.send.duration", 5, nsAttrsSuccess);
+        assertCounterValue(metrics, "pulsar.client.producer.message.send.size", "hello".length() * 5, nsAttrs);
+
+
+        assertCounterValue(metrics, "pulsar.client.producer.opened", 1, nsAttrs);
+
+        producer.close();
+        client.close();
+
+        metrics = collectMetrics();
+        assertCounterValue(metrics, "pulsar.client.producer.closed", 1, nsAttrs);
+        assertCounterValue(metrics, "pulsar.client.connections.closed", 1, Attributes.empty());
+    }
+
+    @Test
+    public void testConnectionsFailedMetrics() throws Exception {
+        String topic = newTopicName();
+
+        @Cleanup
+        PulsarClient client = PulsarClient.builder()
+                .serviceUrl("pulsar://invalid-pulsar-address:1234")
+                .operationTimeout(3, TimeUnit.SECONDS)
+                .openTelemetry(otel)
+                .build();
+
+        try {
+            client.newProducer(Schema.STRING)
+                    .topic(topic)
+                    .create();
+            fail("Should have failed the producer creation");
+        } catch (Exception e) {
+            // Expected
+        }
+
+        var metrics = collectMetrics();
+
+        assertTrue(getCounterValue(metrics, "pulsar.client.connections.failed",
+                Attributes.builder().put("pulsar.failure.type", "tcp-failed").build()) >= 1);
+    }
+
+    @Test
+    public void testPublishFailedMetrics() throws Exception {
+        String topic = newTopicName();
+
+        @Cleanup
+        PulsarClient client = PulsarClient.builder()
+                .serviceUrl(admin.getServiceUrl())
+                .operationTimeout(3, TimeUnit.SECONDS)
+                .openTelemetry(otel)
+                .build();
+
+        @Cleanup
+        Producer<String> producer = client.newProducer(Schema.STRING)
+                .topic(topic)
+                .sendTimeout(3, TimeUnit.SECONDS)
+                .create();
+
+        // Make the client switch to non-existing broker to make publish fail
+        client.updateServiceUrl("pulsar://invalid-address:6650");
+
+
+        try {
+            producer.send("Hello");
+            fail("Should have failed to publish");
+        } catch (Exception e) {
+            // expected
+        }
+
+        var metrics = collectMetrics();
+
+        Attributes nsAttrs = Attributes.builder()
+                .put("pulsar.tenant", "my-property")
+                .put("pulsar.namespace", "my-property/my-ns")
+                .build();
+        Attributes nsAttrsFailure = nsAttrs.toBuilder()
+                .put("success", false)
+                .build();
+
+        assertCounterValue(metrics, "pulsar.client.producer.message.pending.count", 0, nsAttrs);
+        assertCounterValue(metrics, "pulsar.client.producer.message.pending.size", 0, nsAttrs);
+        assertHistoCountValue(metrics, "pulsar.client.producer.message.send.duration", 1, nsAttrsFailure);
+        assertHistoCountValue(metrics, "pulsar.client.producer.rpc.send.duration", 1, nsAttrsFailure);
+    }
+
+    @Test
+    public void testConsumerMetrics() throws Exception {
+        String topic = newTopicName();
+
+        PulsarClient client = PulsarClient.builder()
+                .serviceUrl(pulsar.getBrokerServiceUrl())
+                .openTelemetry(otel)
+                .build();
+
+        @Cleanup
+        Producer<String> producer = client.newProducer(Schema.STRING)
+                .topic(topic)
+                .create();
+
+        Consumer<String> consumer = client.newConsumer(Schema.STRING)
+                .topic(topic)
+                .subscriptionName("my-sub")
+                .ackTimeout(1, TimeUnit.SECONDS)
+                .subscriptionType(SubscriptionType.Shared)
+                .subscribe();
+
+        for (int i = 0; i < 10; i++) {
+            producer.send("Hello");
+        }
+
+        Thread.sleep(1000);
+
+        Attributes nsAttrs = Attributes.builder()
+                .put("pulsar.tenant", "my-property")
+                .put("pulsar.namespace", "my-property/my-ns")
+                .build();
+        var metrics = collectMetrics();
+
+        assertCounterValue(metrics, "pulsar.client.connections.opened", 1, Attributes.empty());
+
+        assertHistoCountValue(metrics, "pulsar.client.lookup.duration", 2,
+                Attributes.builder()
+                        .put("pulsar.lookup.transport-type", "binary")
+                        .put("pulsar.lookup.type", "topic")
+                        .put("success", true)
+                        .build());
+        assertHistoCountValue(metrics, "pulsar.client.lookup.duration", 2,
+                Attributes.builder()
+                        .put("pulsar.lookup.transport-type", "binary")
+                        .put("pulsar.lookup.type", "metadata")
+                        .put("success", true)
+                        .build());
+
+        assertCounterValue(metrics, "pulsar.client.consumer.prefetched.count", 10, nsAttrs);
+        assertCounterValue(metrics, "pulsar.client.consumer.prefetched.size", "hello".length() * 10, nsAttrs);
+        assertCounterValue(metrics, "pulsar.client.consumer.opened", 1, nsAttrs);
+
+        Message<String> msg1 = consumer.receive();
+        consumer.acknowledge(msg1);
+
+        Message<String> msg2 = consumer.receive();
+        consumer.negativeAcknowledge(msg2);
+
+        /* Message<String> msg3 = */ consumer.receive();
+
+        metrics = collectMetrics();
+        assertCounterValue(metrics, "pulsar.client.consumer.prefetched.count", 7, nsAttrs);
+        assertCounterValue(metrics, "pulsar.client.consumer.prefetched.size", "hello".length() * 7, nsAttrs);
+
+        // Let msg3 to reach ack-timeout
+        Thread.sleep(3000);
+
+        metrics = collectMetrics();
+        assertCounterValue(metrics, "pulsar.client.consumer.prefetched.count", 8, nsAttrs);
+        assertCounterValue(metrics, "pulsar.client.consumer.prefetched.size", "hello".length() * 8, nsAttrs);
+
+        assertCounterValue(metrics, "pulsar.client.consumer.message.ack", 1, nsAttrs);
+        assertCounterValue(metrics, "pulsar.client.consumer.message.nack", 1, nsAttrs);
+        assertCounterValue(metrics, "pulsar.client.consumer.message.ack.timeout", 1, nsAttrs);
+
+        client.close();
+
+        metrics = collectMetrics();
+        assertCounterValue(metrics, "pulsar.client.consumer.closed", 1, nsAttrs);
+        assertCounterValue(metrics, "pulsar.client.connections.closed", 1, Attributes.empty());
+    }
+}

pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java

@@ -159,12 +159,6 @@ public CompletableFuture<PartitionedTopicMetadata> getPartitionedTopicMetadata(T
                 CompletableFuture<PartitionedTopicMetadata> newFuture =
                         getPartitionedTopicMetadata(serviceNameResolver.resolveHost(), topicName);
                 newFutureCreated.setValue(newFuture);
-                newFuture.thenRun(() -> {
-                    histoGetTopicMetadata.recordSuccess(System.nanoTime() - startTime);
-                }).exceptionally(x -> {
-                    histoGetTopicMetadata.recordFailure(System.nanoTime() - startTime);
-                    return null;
-                });
                 return newFuture;
             });
         } finally {

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

@@ -420,8 +420,8 @@ protected ConsumerImpl(PulsarClientImpl client, String topic, ConsumerConfigurat
                 "The number of acknowledged messages", topic, Attributes.empty());
         consumerNacksCounter = ip.newCounter("pulsar.client.consumer.message.nack", Unit.Messages,
                 "The number of negatively acknowledged messages", topic, Attributes.empty());
-        consumerDlqMessagesCounter = ip.newCounter("pulsar.client.consumer.dlq", Unit.Messages,
-                "Number of messages sent to DLQ", topic, Attributes.empty());
+        consumerDlqMessagesCounter = ip.newCounter("pulsar.client.consumer.message.dlq", Unit.Messages,
+                "The number of messages sent to DLQ", topic, Attributes.empty());
         grabCnx();
 
         consumersOpenedCounter.increment();

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java

@@ -291,15 +291,15 @@ public ProducerImpl(PulsarClientImpl client, String topic, ProducerConfiguration
                 "Publish RPC latency experienced internally by the client when sending data to receiving an ack", topic,
                 Attributes.empty());
         publishedBytesCounter = ip.newCounter("pulsar.client.producer.message.send.size",
-                Unit.Bytes, "Bytes published", topic, Attributes.empty());
+                Unit.Bytes, "The number of bytes published", topic, Attributes.empty());
         pendingMessagesCounter = ip.newUpDownCounter("pulsar.client.producer.message.pending.count", Unit.Messages,
                 "Pending messages for this producer", topic, Attributes.empty());
         pendingBytesCounter = ip.newUpDownCounter("pulsar.client.producer.message.pending.size", Unit.Bytes,
                 "Pending bytes for this producer", topic, Attributes.empty());
         producersOpenedCounter = ip.newCounter("pulsar.client.producer.opened", Unit.Sessions,
-                "Counter of producer sessions opened", topic, Attributes.empty());
+                "The number of producer sessions opened", topic, Attributes.empty());
         producersClosedCounter = ip.newCounter("pulsar.client.producer.closed", Unit.Sessions,
-                "Counter of producer sessions closed", topic, Attributes.empty());
+                "The number of producer sessions closed", topic, Attributes.empty());
 
         this.connectionHandler = new ConnectionHandler(this,
                 new BackoffBuilder()
@@ -2145,6 +2145,7 @@ private void failPendingMessages(ClientCnx cnx, PulsarClientException ex) {
                     log.warn("[{}] [{}] Got exception while completing the callback for msg {}:", topic, producerName,
                             op.sequenceId, t);
                 }
+
                 client.getMemoryLimitController().releaseMemory(op.uncompressedSize);
                 ReferenceCountUtil.safeRelease(op.cmd);
                 op.recycle();