Fixed tests code

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

@@ -263,8 +263,8 @@ public OpSendMsg createOpSendMsg() throws IOException {
             // Because when invoke `ProducerImpl.processOpSendMsg` on flush,
             // if `op.msg != null && isBatchMessagingEnabled()` checks true, it will call `batchMessageAndSend` to flush
             // messageContainers before publishing this one-batch message.
-            op = OpSendMsg.create(producer, messages, cmd, messageMetadata.getSequenceId(), firstCallback,
-                    batchAllocatedSizeBytes);
+            op = OpSendMsg.create(producer.rpcLatencyHistogram, messages, cmd, messageMetadata.getSequenceId(),
+                    firstCallback, batchAllocatedSizeBytes);
 
             // NumMessagesInBatch and BatchSizeByte will not be serialized to the binary cmd. It's just useful for the
             // ProducerStats
@@ -314,7 +314,7 @@ public OpSendMsg createOpSendMsg() throws IOException {
                     messageMetadata.getUncompressedSize(), encryptedPayload.readableBytes());
         }
 
-        OpSendMsg op = OpSendMsg.create(producer, messages, cmd, messageMetadata.getSequenceId(),
+        OpSendMsg op = OpSendMsg.create(producer.rpcLatencyHistogram, messages, cmd, messageMetadata.getSequenceId(),
                 messageMetadata.getHighestSequenceId(), firstCallback, batchAllocatedSizeBytes);
 
         op.setNumMessagesInBatch(numMessagesInBatch);

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

@@ -40,7 +40,6 @@
 import io.netty.util.Timeout;
 import io.netty.util.TimerTask;
 import io.netty.util.concurrent.ScheduledFuture;
-import io.opentelemetry.api.common.AttributeKey;
 import io.opentelemetry.api.common.Attributes;
 import java.io.IOException;
 import java.net.URI;
@@ -179,7 +178,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
     private boolean errorState;
 
     private final LatencyHistogram latencyHistogram;
-    private final LatencyHistogram rpcLatencyHistogram;
+    final LatencyHistogram rpcLatencyHistogram;
     private final Counter publishedBytesCounter;
     private final UpDownCounter pendingMessagesCounter;
     private final UpDownCounter pendingBytesCounter;
@@ -745,9 +744,9 @@ private void serializeAndSendMessage(MessageImpl<?> msg,
             if (msg.getSchemaState() == MessageImpl.SchemaState.Ready) {
                 ByteBufPair cmd = sendMessage(producerId, sequenceId, numMessages, messageId, msgMetadata,
                         encryptedPayload);
-                op = OpSendMsg.create(this, msg, cmd, sequenceId, callback);
+                op = OpSendMsg.create(rpcLatencyHistogram, msg, cmd, sequenceId, callback);
             } else {
-                op = OpSendMsg.create(this, msg, null, sequenceId, callback);
+                op = OpSendMsg.create(rpcLatencyHistogram, msg, null, sequenceId, callback);
                 final MessageMetadata finalMsgMetadata = msgMetadata;
                 op.rePopulate = () -> {
                     if (msgMetadata.hasChunkId()) {
@@ -1455,7 +1454,7 @@ public ReferenceCounted touch(Object hint) {
     }
 
     protected static final class OpSendMsg {
-        ProducerImpl<?> producer;
+        LatencyHistogram rpcLatencyHistogram;
         MessageImpl<?> msg;
         List<MessageImpl<?>> msgs;
         ByteBufPair cmd;
@@ -1475,7 +1474,7 @@ protected static final class OpSendMsg {
         int chunkId = -1;
 
         void initialize() {
-            producer = null;
+            rpcLatencyHistogram = null;
             msg = null;
             msgs = null;
             cmd = null;
@@ -1495,11 +1494,11 @@ void initialize() {
             chunkedMessageCtx = null;
         }
 
-        static OpSendMsg create(ProducerImpl<?> producer, MessageImpl<?> msg, ByteBufPair cmd, long sequenceId,
+        static OpSendMsg create(LatencyHistogram rpcLatencyHistogram, MessageImpl<?> msg, ByteBufPair cmd, long sequenceId,
                                 SendCallback callback) {
             OpSendMsg op = RECYCLER.get();
             op.initialize();
-            op.producer = producer;
+            op.rpcLatencyHistogram = rpcLatencyHistogram;
             op.msg = msg;
             op.cmd = cmd;
             op.callback = callback;
@@ -1509,12 +1508,12 @@ static OpSendMsg create(ProducerImpl<?> producer, MessageImpl<?> msg, ByteBufPai
             return op;
         }
 
-        static OpSendMsg create(ProducerImpl<?> producer, List<MessageImpl<?>> msgs, ByteBufPair cmd, long sequenceId,
+        static OpSendMsg create(LatencyHistogram rpcLatencyHistogram, List<MessageImpl<?>> msgs, ByteBufPair cmd, long sequenceId,
                                 SendCallback callback,
                                 int batchAllocatedSize) {
             OpSendMsg op = RECYCLER.get();
             op.initialize();
-            op.producer = producer;
+            op.rpcLatencyHistogram = rpcLatencyHistogram;
             op.msgs = msgs;
             op.cmd = cmd;
             op.callback = callback;
@@ -1528,12 +1527,12 @@ static OpSendMsg create(ProducerImpl<?> producer, List<MessageImpl<?>> msgs, Byt
             return op;
         }
 
-        static OpSendMsg create(ProducerImpl<?> producer, List<MessageImpl<?>> msgs, ByteBufPair cmd,
+        static OpSendMsg create(LatencyHistogram rpcLatencyHistogram, List<MessageImpl<?>> msgs, ByteBufPair cmd,
                                 long lowestSequenceId,
                                 long highestSequenceId, SendCallback callback, int batchAllocatedSize) {
             OpSendMsg op = RECYCLER.get();
             op.initialize();
-            op.producer = producer;
+            op.rpcLatencyHistogram = rpcLatencyHistogram;
             op.msgs = msgs;
             op.cmd = cmd;
             op.callback = callback;
@@ -1585,9 +1584,9 @@ void sendComplete(final Exception e) {
                 }
 
                 if (e == null) {
-                    producer.rpcLatencyHistogram.recordSuccess(now - this.lastSentAt);
+                    rpcLatencyHistogram.recordSuccess(now - this.lastSentAt);
                 } else {
-                    producer.rpcLatencyHistogram.recordFailure(now - this.lastSentAt);
+                    rpcLatencyHistogram.recordFailure(now - this.lastSentAt);
                 }
 
                 callback.sendComplete(finalEx);

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

@@ -32,12 +32,20 @@
 
 public class LatencyHistogram {
 
+    // Used for tests
+    public final static LatencyHistogram NOOP = new LatencyHistogram(null, null, null, null, null) {
+        public void recordSuccess(long latencyNanos) {
+        }
+
+        public void recordFailure(long latencyNanos) {
+        }
+    };
+
     private static final List<Double> latencyHistogramBuckets =
             Lists.newArrayList(.0005, .001, .0025, .005, .01, .025, .05, .1, .25, .5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0);
 
     private static final double NANOS = TimeUnit.SECONDS.toNanos(1);
 
-
     private final Attributes successAttributes;
 
     private final Attributes failedAttributes;

pulsar-client/src/test/java/org/apache/pulsar/client/impl/OpSendMsgQueueTest.java

@@ -23,6 +23,7 @@
 import static org.testng.Assert.assertEquals;
 import com.google.common.collect.Lists;
 import java.util.Arrays;
+import org.apache.pulsar.client.impl.metrics.LatencyHistogram;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
 
@@ -39,7 +40,7 @@ public void createMockMessage() {
     }
 
     private ProducerImpl.OpSendMsg createDummyOpSendMsg() {
-        return ProducerImpl.OpSendMsg.create(message, null, 0L, null);
+        return ProducerImpl.OpSendMsg.create(LatencyHistogram.NOOP, message, null, 0L, null);
     }
 
     @Test

pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java

@@ -27,6 +27,7 @@
 import static org.testng.Assert.assertTrue;
 import java.nio.ByteBuffer;
 import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.metrics.LatencyHistogram;
 import org.apache.pulsar.common.api.proto.MessageMetadata;
 import org.mockito.Mockito;
 import org.testng.annotations.Test;
@@ -42,6 +43,7 @@ public void testChunkedMessageCtxDeallocate() {
         for (int i = 0; i < totalChunks; i++) {
             ProducerImpl.OpSendMsg opSendMsg =
                     ProducerImpl.OpSendMsg.create(
+                            LatencyHistogram.NOOP,
                             MessageImpl.create(new MessageMetadata(), ByteBuffer.allocate(0), Schema.STRING, null),
                             null, 0, null);
             opSendMsg.chunkedMessageCtx = ctx;

pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java

@@ -53,6 +53,7 @@
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
 import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
 import org.apache.pulsar.client.util.ExecutorProvider;
 import org.apache.pulsar.client.util.ScheduledExecutorProvider;
 import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace;
@@ -180,7 +181,7 @@ public void testInitializeWithTimer() throws PulsarClientException {
         ClientConfigurationData conf = new ClientConfigurationData();
         @Cleanup("shutdownGracefully")
         EventLoopGroup eventLoop = EventLoopUtil.newEventLoopGroup(1, false, new DefaultThreadFactory("test"));
-        ConnectionPool pool = Mockito.spy(new ConnectionPool(conf, eventLoop));
+        ConnectionPool pool = Mockito.spy(new ConnectionPool(new InstrumentProvider(conf), conf, eventLoop));
         conf.setServiceUrl("pulsar://localhost:6650");
 
         HashedWheelTimer timer = new HashedWheelTimer();
@@ -205,7 +206,7 @@ public void testResourceCleanup() throws Exception {
         ClientConfigurationData conf = new ClientConfigurationData();
         conf.setServiceUrl("");
         initializeEventLoopGroup(conf);
-        try (ConnectionPool connectionPool = new ConnectionPool(conf, eventLoopGroup)) {
+        try (ConnectionPool connectionPool = new ConnectionPool(new InstrumentProvider(conf), conf, eventLoopGroup)) {
             assertThrows(() -> new PulsarClientImpl(conf, eventLoopGroup, connectionPool));
         } finally {
             // Externally passed eventLoopGroup should not be shutdown.