Addressed comments

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

@@ -93,20 +93,20 @@ public BinaryProtoLookupService(PulsarClientImpl client,
         this.listenerName = listenerName;
         updateServiceUrl(serviceUrl);
 
-        Attributes attrs = Attributes.of(AttributeKey.stringKey("transport-type"), "binary");
+        Attributes attrs = Attributes.of(AttributeKey.stringKey("pulsar.lookup.transport-type"), "binary");
 
         histoGetBroker = client.instrumentProvider().newLatencyHistogram("pulsar.client.lookup",
-                "Lookup operations",
-                attrs.toBuilder().put("type", "topic").build());
+                "Duration of lookup operations",
+                attrs.toBuilder().put("pulsar.lookup.type", "topic").build());
         histoGetTopicMetadata = client.instrumentProvider().newLatencyHistogram("pulsar.client.lookup",
-                "Lookup operations",
-                attrs.toBuilder().put("type", "metadata").build());
+                "Duration of lookup operations",
+                attrs.toBuilder().put("pulsar.lookup.type", "metadata").build());
         histoGetSchema = client.instrumentProvider().newLatencyHistogram("pulsar.client.lookup",
-                "Lookup operations",
-                attrs.toBuilder().put("type", "schema").build());
+                "Duration of lookup operations",
+                attrs.toBuilder().put("pulsar.lookup.type", "schema").build());
         histoListTopics = client.instrumentProvider().newLatencyHistogram("pulsar.client.lookup",
-                "Lookup operations",
-                attrs.toBuilder().put("type", "list-topics").build());
+                "Duration of lookup operations",
+                attrs.toBuilder().put("pulsar.lookup.type", "list-topics").build());
     }
 
     @Override
@@ -160,9 +160,9 @@ public CompletableFuture<PartitionedTopicMetadata> getPartitionedTopicMetadata(T
                         getPartitionedTopicMetadata(serviceNameResolver.resolveHost(), topicName);
                 newFutureCreated.setValue(newFuture);
                 newFuture.thenRun(() -> {
-                    histoGetBroker.recordSuccess(System.nanoTime() - startTime);
+                    histoGetTopicMetadata.recordSuccess(System.nanoTime() - startTime);
                 }).exceptionally(x -> {
-                    histoGetBroker.recordFailure(System.nanoTime() - startTime);
+                    histoGetTopicMetadata.recordFailure(System.nanoTime() - startTime);
                     return null;
                 });
                 return newFuture;

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

@@ -271,9 +271,9 @@ public ClientCnx(InstrumentProvider instrumentProvider, ClientConfigurationData
         this.clientVersion = "Pulsar-Java-v" + PulsarVersion.getVersion()
                 + (conf.getDescription() == null ? "" : ("-" + conf.getDescription()));
         this.connectionsOpenedCounter = instrumentProvider.newCounter("pulsar.client.connections.opened", Unit.Connections,
-                "Counter of connections opened", Attributes.empty());
+                "The number of connections opened", Attributes.empty());
         this.connectionsClosedCounter = instrumentProvider.newCounter("pulsar.client.connections.closed", Unit.Connections,
-                "Counter of connections closed", Attributes.empty());
+                "The number of connections closed", Attributes.empty());
 
     }
 

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

@@ -157,11 +157,12 @@ public ConnectionPool(InstrumentProvider instrumentProvider,
             }, idleDetectionIntervalSeconds, idleDetectionIntervalSeconds, TimeUnit.SECONDS);
         }
 
-        connectionsTcpFailureCounter = instrumentProvider.newCounter("pulsar.client.connections.failed", Unit.None,
-                "Counter of failed connections", Attributes.builder().put("type", "tcp-failed").build());
+        connectionsTcpFailureCounter =
+                instrumentProvider.newCounter("pulsar.client.connections.failed", Unit.Connections,
+                        "The number of failed connection attempts", Attributes.builder().put("pulsar.failure.type", "tcp-failed").build());
         connectionsHandshakeFailureCounter = instrumentProvider.newCounter("pulsar.client.connections.failed",
-                Unit.None, "Counter of failed connections",
-                Attributes.builder().put("type", "handshake").build());
+                Unit.Connections, "The number of failed connection attempts",
+                Attributes.builder().put("pulsar.failure.type", "handshake").build());
     }
 
     private static AddressResolver<InetSocketAddress> createAddressResolver(ClientConfigurationData conf,

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

@@ -1228,10 +1228,7 @@ public int getTotalIncomingMessages() {
 
     protected void clearIncomingMessages() {
         // release messages if they are pooled messages
-        incomingMessages.forEach(msg -> {
-
-            msg.release();
-        });
+        incomingMessages.forEach(Message::release);
         incomingMessages.clear();
         resetIncomingMessageSize();
     }

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

@@ -408,19 +408,19 @@ protected ConsumerImpl(PulsarClientImpl client, String topic, ConsumerConfigurat
                 "Counter of sessions opened", attrs.toBuilder().put("type", "consumer").build());
         consumersClosedCounter = ip.newCounter("pulsar.client.session.closed", Unit.Sessions,
                 "Counter of sessions closed", attrs.toBuilder().put("type", "consumer").build());
-        messagesReceivedCounter = ip.newCounter("pulsar.client.received", Unit.Messages,
-                "Number of messages received", attrs);
-        bytesReceivedCounter = ip.newCounter("pulsar.client.received", Unit.Bytes,
-                "Bytes received", attrs);
-        messagesPrefetchedGauge = ip.newUpDownCounter("pulsar.client.consumer.preteched.messages", Unit.Messages,
+        messagesReceivedCounter = ip.newCounter("pulsar.client.received.count", Unit.Messages,
+                "The number of messages explicitly received by the consumer application", attrs);
+        bytesReceivedCounter = ip.newCounter("pulsar.client.received.size", Unit.Bytes,
+                "The number of bytes explicitly received by the consumer application", attrs);
+        messagesPrefetchedGauge = ip.newUpDownCounter("pulsar.client.consumer.prefetched.count", Unit.Messages,
                 "Number of messages currently sitting in the consumer pre-fetch queue", attrs);
-        bytesPrefetchedGauge = ip.newUpDownCounter("pulsar.client.consumer.preteched", Unit.Bytes,
+        bytesPrefetchedGauge = ip.newUpDownCounter("pulsar.client.consumer.prefetched.size", Unit.Bytes,
                 "Total number of bytes currently sitting in the consumer pre-fetch queue", attrs);
 
-        consumerAcksCounter = ip.newCounter("pulsar.client.consumer.ack", Unit.Messages,
-                "Number of ack operations", attrs);
-        consumerNacksCounter = ip.newCounter("pulsar.client.consumer.nack", Unit.Messages,
-                "Number of negative ack operations", attrs);
+        consumerAcksCounter = ip.newCounter("pulsar.client.consumer.message.ack", Unit.Messages,
+                "The number of acknowledged messages", attrs);
+        consumerNacksCounter = ip.newCounter("pulsar.client.consumer.message.nack", Unit.Messages,
+                "The number of negatively acknowledged messages", attrs);
         consumerDlqMessagesCounter = ip.newCounter("pulsar.client.consumer.dlq", Unit.Messages,
                 "Number of messages sent to DLQ", attrs);
         grabCnx();

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

@@ -75,20 +75,20 @@ public HttpLookupService(InstrumentProvider instrumentProvider, ClientConfigurat
         this.useTls = conf.isUseTls();
         this.listenerName = conf.getListenerName();
 
-        Attributes attrs = Attributes.of(AttributeKey.stringKey("transport-type"), "binary");
+        Attributes attrs = Attributes.of(AttributeKey.stringKey("pulsar.lookup.transport-type"), "binary");
 
         histoGetBroker = instrumentProvider.newLatencyHistogram("pulsar.client.lookup",
-                "Lookup operations",
-                attrs.toBuilder().put("type", "topic").build());
+                "Duration of lookup operations",
+                attrs.toBuilder().put("pulsar.lookup.type", "topic").build());
         histoGetTopicMetadata = instrumentProvider.newLatencyHistogram("pulsar.client.lookup",
-                "Lookup operations",
-                attrs.toBuilder().put("type", "metadata").build());
+                "Duration of lookup operations",
+                attrs.toBuilder().put("pulsar.lookup.type", "metadata").build());
         histoGetSchema = instrumentProvider.newLatencyHistogram("pulsar.client.lookup",
-                "Lookup operations",
-                attrs.toBuilder().put("type", "schema").build());
+                "Duration of lookup operations",
+                attrs.toBuilder().put("pulsar.lookup.type", "schema").build());
         histoListTopics = instrumentProvider.newLatencyHistogram("pulsar.client.lookup",
-                "Lookup operations",
-                attrs.toBuilder().put("type", "list-topics").build());
+                "Duration of lookup operations",
+                attrs.toBuilder().put("pulsar.lookup.type", "list-topics").build());
     }
 
     @Override

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

@@ -285,15 +285,15 @@ public ProducerImpl(PulsarClientImpl client, String topic, ProducerConfiguration
 
         InstrumentProvider ip = client.instrumentProvider();
         Attributes attrs = ip.getAttributes(topic);
-        latencyHistogram = ip.newLatencyHistogram("pulsar.client.producer.latency",
+        latencyHistogram = ip.newLatencyHistogram("pulsar.client.producer.message.send.duration",
                 "Publish latency experienced by the application, includes client batching time", attrs);
-        rpcLatencyHistogram = ip.newLatencyHistogram("pulsar.client.producer.rpc.latency",
+        rpcLatencyHistogram = ip.newLatencyHistogram("pulsar.client.producer.rpc.send.duration",
                 "Publish RPC latency experienced internally by the client when sending data to receiving an ack", attrs);
-        publishedBytesCounter = ip.newCounter("pulsar.client.producer.published",
+        publishedBytesCounter = ip.newCounter("pulsar.client.producer.message.send.size",
                 Unit.Bytes, "Bytes published", attrs);
-        pendingMessagesCounter = ip.newUpDownCounter("pulsar.client.producer.pending.messages.count", Unit.Messages,
+        pendingMessagesCounter = ip.newUpDownCounter("pulsar.client.producer.message.pending.count", Unit.Messages,
                 "Pending messages for this producer", attrs);
-        pendingBytesCounter = ip.newUpDownCounter("pulsar.client.producer.pending.count", Unit.Bytes,
+        pendingBytesCounter = ip.newUpDownCounter("pulsar.client.producer.message.pending.size", Unit.Bytes,
                 "Pending bytes for this producer", attrs);
         producersOpenedCounter = ip.newCounter("pulsar.client.session.opened", Unit.Sessions,
                 "Counter of sessions opened", attrs.toBuilder().put("type", "producer").build());

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

@@ -124,7 +124,7 @@ public UnAckedMessageTracker(PulsarClientImpl client, ConsumerBase<?> consumerBa
 
         InstrumentProvider ip = client.instrumentProvider();
         Attributes attrs = ip.getAttributes(consumerBase.getTopic());
-        consumerAckTimeoutsCounter = ip.newCounter("pulsar.client.consumer.ack.timeout", Unit.Messages,
+        consumerAckTimeoutsCounter = ip.newCounter("pulsar.client.consumer.message.ack.timeout", Unit.Messages,
                 "Number of ack timeouts events", attrs);
 
         if (conf.getAckTimeoutRedeliveryBackoff() == null) {

pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java

@@ -397,7 +397,6 @@ public class ClientConfigurationData implements Serializable, Cloneable {
     )
     private String description;
 
-
     private transient OpenTelemetry openTelemetry;
 
     private MetricsCardinality openTelemetryMetricsCardinality = MetricsCardinality.Topic;

pulsar-client/src/main/java/org/apache/pulsar/client/impl/metrics/InstrumentProvider.java

@@ -1,9 +1,11 @@
 package org.apache.pulsar.client.impl.metrics;
 
+import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.api.common.AttributesBuilder;
 import io.opentelemetry.api.metrics.Meter;
+import org.apache.pulsar.PulsarVersion;
 import org.apache.pulsar.client.api.MetricsCardinality;
 import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
 import org.apache.pulsar.common.naming.TopicName;
@@ -14,7 +16,17 @@ public class InstrumentProvider {
     private final MetricsCardinality metricsCardinality;
 
     public InstrumentProvider(ClientConfigurationData conf) {
-        this.meter = conf.getOpenTelemetry().getMeter("org.apache.pulsar.client");
+        OpenTelemetry otel = conf.getOpenTelemetry();
+        if (otel == null) {
+            // By default, metrics are disabled, unless the OTel java agent is configured.
+            // This allows to enable metrics without any code change.
+            otel = GlobalOpenTelemetry.get();
+        }
+
+        this.meter = otel.getMeterProvider()
+                .meterBuilder("org.apache.pulsar.client")
+                .setInstrumentationVersion(PulsarVersion.getVersion())
+                .build();
         this.metricsCardinality = conf.getOpenTelemetryMetricsCardinality();
     }
 
@@ -29,17 +41,17 @@ public Attributes getAttributes(String topic) {
         switch (metricsCardinality) {
             case Partition:
                 if (tn.isPartitioned()) {
-                    ab.put("partition", tn.getPartitionIndex());
+                    ab.put("pulsar.partition", tn.getPartitionIndex());
                 }
                 // fallthrough
             case Topic:
-                ab.put("topic", tn.getPartitionedTopicName());
+                ab.put("pulsar.topic", tn.getPartitionedTopicName());
                 // fallthrough
             case Namespace:
-                ab.put("namespace", tn.getNamespace());
+                ab.put("pulsar.namespace", tn.getNamespace());
                 // fallthrough
             case Tenant:
-                ab.put("tenant", tn.getTenant());
+                ab.put("pulsar.tenant", tn.getTenant());
         }
 
         return ab.build();