[fix][client] Fix incorrect producer.getPendingQueueSize due to incomplete queue implementation (#24184)

Co-authored-by: Lari Hotari <[email protected]>
Co-authored-by: Lari Hotari <[email protected]>

Author: poorbarcode

pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java

@@ -48,6 +48,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import lombok.Cleanup;
@@ -66,6 +67,7 @@
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Schema;
@@ -77,6 +79,7 @@
 import org.apache.pulsar.client.api.schema.SchemaDefinition;
 import org.apache.pulsar.client.impl.ConsumerImpl;
 import org.apache.pulsar.client.impl.MultiTopicsConsumerImpl;
+import org.apache.pulsar.client.impl.ProducerImpl;
 import org.apache.pulsar.client.impl.schema.KeyValueSchemaImpl;
 import org.apache.pulsar.client.impl.schema.ProtobufSchema;
 import org.apache.pulsar.client.impl.schema.SchemaInfoImpl;
@@ -96,6 +99,7 @@
 import org.apache.pulsar.metadata.api.MetadataCache;
 import org.apache.pulsar.metadata.api.MetadataSerde;
 import org.apache.pulsar.metadata.api.Stat;
+import org.awaitility.Awaitility;
 import org.testng.Assert;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
@@ -1454,9 +1458,9 @@ public User(String name) {
     }
 
     /**
-     * This test validates that consumer/producers should recover on topic whose 
+     * This test validates that consumer/producers should recover on topic whose
      * schema ledgers are not able to open due to non-recoverable error.
-     * 
+     *
      * @throws Exception
      */
     @Test
@@ -1524,6 +1528,31 @@ public SchemaStorageFormat.SchemaLocator deserialize(String path, byte[] content
         producer.close();
     }
 
+    @Test
+    public void testPendingQueueSizeIfIncompatible() throws Exception {
+        final String namespace = BrokerTestUtil.newUniqueName(PUBLIC_TENANT + "/ns");
+        admin.namespaces().createNamespace(namespace, Sets.newHashSet(CLUSTER_NAME));
+        admin.namespaces().setSchemaCompatibilityStrategy(namespace, SchemaCompatibilityStrategy.ALWAYS_INCOMPATIBLE);
+        final String topic = BrokerTestUtil.newUniqueName(namespace + "/tp");
+        admin.topics().createNonPartitionedTopic(topic);
+
+        ProducerImpl producer = (ProducerImpl) pulsarClient.newProducer(Schema.AUTO_PRODUCE_BYTES())
+                .maxPendingMessages(50).enableBatching(false).topic(topic).create();
+        producer.newMessage(Schema.STRING).value("msg").sendAsync();
+        AtomicReference<CompletableFuture<MessageId>> latestSend = new AtomicReference<>();
+        for (int i = 0; i < 100; i++) {
+            latestSend.set(producer.newMessage(Schema.BOOL).value(false).sendAsync());
+        }
+        Awaitility.await().untilAsserted(() -> {
+            assertTrue(latestSend.get().isDone());
+            assertEquals(producer.getPendingQueueSize(), 0);
+        });
+
+        // cleanup.
+        producer.close();
+        admin.topics().delete(topic, false);
+    }
+
     @Test
     public void testTopicSchemaMetadata() throws Exception {
         final String tenant = PUBLIC_TENANT;

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

@@ -1734,7 +1734,8 @@ protected OpSendMsg newObject(Handle<OpSendMsg> handle) {
      * This queue is not thread safe.
      */
     protected static class OpSendMsgQueue implements Iterable<OpSendMsg> {
-        private final Queue<OpSendMsg> delegate = new ArrayDeque<>();
+        @VisibleForTesting
+        final Queue<OpSendMsg> delegate = new ArrayDeque<>();
         private int forEachDepth = 0;
         private List<OpSendMsg> postponedOpSendMgs;
         private final AtomicInteger messagesCount = new AtomicInteger(0);
@@ -1791,7 +1792,35 @@ public int messagesCount() {
 
         @Override
         public Iterator<OpSendMsg> iterator() {
-            return delegate.iterator();
+            Iterator<OpSendMsg> delegateIterator = delegate.iterator();
+            return new Iterator<OpSendMsg>() {
+                OpSendMsg currentOp;
+
+                @Override
+                public boolean hasNext() {
+                    return delegateIterator.hasNext();
+                }
+
+                @Override
+                public OpSendMsg next() {
+                    currentOp = delegateIterator.next();
+                    return currentOp;
+                }
+
+                @Override
+                public void remove() {
+                    delegateIterator.remove();
+                    if (currentOp != null) {
+                       messagesCount.addAndGet(-currentOp.numMessagesInBatch);
+                       currentOp = null;
+                     }
+                }
+
+                @Override
+                public void forEachRemaining(Consumer<? super OpSendMsg> action) {
+                    delegateIterator.forEachRemaining(action);
+                }
+            };
         }
     }
 

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 java.util.Iterator;
 import org.apache.pulsar.client.impl.metrics.LatencyHistogram;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
@@ -83,4 +84,21 @@ public void shouldPostponeAddsAlsoInRecursiveCalls() {
         // then
         assertEquals(Lists.newArrayList(queue), Arrays.asList(opSendMsg, opSendMsg2, opSendMsg3, opSendMsg4));
     }
+
+    @Test
+    public void testIteratorRemove() {
+        ProducerImpl.OpSendMsgQueue queue = new ProducerImpl.OpSendMsgQueue();
+        for (int i = 0; i < 10; i++) {
+            queue.add(createDummyOpSendMsg());
+        }
+
+        Iterator<ProducerImpl.OpSendMsg> iterator = queue.iterator();
+        while (iterator.hasNext()) {
+            iterator.next();
+            iterator.remove();
+        }
+        // Verify: the result of "messagesCount()" is 0 after removed all items.
+        assertEquals(queue.delegate.size(), 0);
+        assertEquals(queue.messagesCount(), 0);
+    }
 }