[fix][client] Fix producer publishing getting stuck after message with incompatible schema is discarded (apache#24282)

Fixes apache#24262

Main Issue: apache#24262

The issue is a regression of apache#24178. The flow of the issue occurring is as follows

- Publish msg 1, which has an incompatible schema
- Publish msg 2, which has an incompatible schema
- The first message's schema failed to register because it is incompatible
  - The message was discarded.
  - Issue: it did not trigger a schema registration of the following messages

Fix the issue

(cherry picked from commit df2c619)
(cherry picked from commit a66d6b7)

Author: poorbarcode

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

@@ -1528,8 +1528,7 @@ public SchemaStorageFormat.SchemaLocator deserialize(String path, byte[] content
         producer.close();
     }
 
-    // This test fails consistently, disabling until it is fixed. Issue https://github.com/apache/pulsar/issues/24262
-    @Test(enabled = false)
+    @Test
     public void testPendingQueueSizeIfIncompatible() throws Exception {
         final String namespace = BrokerTestUtil.newUniqueName(PUBLIC_TENANT + "/ns");
         admin.namespaces().createNamespace(namespace, Sets.newHashSet(CLUSTER_NAME));
@@ -1538,17 +1537,28 @@ public void testPendingQueueSizeIfIncompatible() throws Exception {
         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();
+                .maxPendingMessages(1000).enableBatching(false).topic(topic).create();
+        producer.newMessage(Schema.STRING).value("msg-1").sendAsync();
         AtomicReference<CompletableFuture<MessageId>> latestSend = new AtomicReference<>();
         for (int i = 0; i < 100; i++) {
-            latestSend.set(producer.newMessage(Schema.BOOL).value(false).sendAsync());
+            final String msg = "msg-with-broken-schema-" + i;
+            latestSend.set(producer.newMessage(Schema.BOOL).value(false).sendAsync().thenApply(v -> {
+                log.info("send complete {}", msg);
+                return null;
+            }).exceptionally(ex -> {
+                log.error("failed to send {}", msg, ex);
+                return null;
+            }));
         }
+        // Verify: msgs with broken schema will be discarded.
         Awaitility.await().untilAsserted(() -> {
             assertTrue(latestSend.get().isDone());
             assertEquals(producer.getPendingQueueSize(), 0);
         });
 
+        // Verify: msgs with compatible schema can be sent successfully.
+        producer.newMessage(Schema.STRING).value("msg-2").sendAsync();
+
         // cleanup.
         producer.close();
         admin.topics().delete(topic, false);