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

Fixes apache#24262

Main Issue: apache#24262

### Motivation

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 

### Modifications

Fix the issue

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);

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

@@ -2464,6 +2464,7 @@ protected void updateLastSeqPushed(OpSendMsg op) {
      *     3-1-1. If {@link #pauseSendingToPreservePublishOrderOnSchemaRegFailure} is true pause all following
      *       publishing to avoid out-of-order issue.
      *     3-1-2. Otherwise, discard the failed message anc continuously publishing the following messages.
+     *            Additionally, the following messages may need schema registration also.
      *   3-2. The new schema registration failed due to other error, retry registering.
      * Note: Since the current method accesses & modifies {@link #pendingMessages}, you should acquire a lock on
      *       {@link ProducerImpl} before calling method.
@@ -2482,6 +2483,7 @@ private void recoverProcessOpSendMsgFrom(ClientCnx cnx, MessageImpl latestMsgAtt
         Iterator<OpSendMsg> msgIterator = pendingMessages.iterator();
         MessageImpl loopStartAt = latestMsgAttemptedRegisteredSchema;
         OpSendMsg loopEndDueToSchemaRegisterNeeded = null;
+        boolean pausedSendingToPreservePublishOrderOnSchemaRegFailure = false;
         while (msgIterator.hasNext()) {
             OpSendMsg op = msgIterator.next();
             if (loopStartAt != null) {
@@ -2526,6 +2528,7 @@ private void recoverProcessOpSendMsgFrom(ClientCnx cnx, MessageImpl latestMsgAtt
                                 + " 2) Unload topic on target cluster. Schema details: {}",
                                 topic, producerName, SchemaUtils.jsonifySchemaInfo(msgSchemaInfo, false));
                         loopEndDueToSchemaRegisterNeeded = op;
+                        pausedSendingToPreservePublishOrderOnSchemaRegFailure = true;
                         break;
                     }
                     // Event 3-1-2.
@@ -2581,23 +2584,27 @@ private void recoverProcessOpSendMsgFrom(ClientCnx cnx, MessageImpl latestMsgAtt
         }
         cnx.ctx().flush();
 
-        // "Event 1-1" or "Event 3-1-1" or "Event 3-2".
+        // "Event 1-1" or "Event 3-1-1" or "Event 3-1-2" or "Event 3-2".
         if (loopEndDueToSchemaRegisterNeeded != null) {
             if (compareAndSetState(State.Connecting, State.Ready)) {
                 // "Event 1-1" happens after "Event 3-1-1".
                 // After a topic unload, ask the producer retry to register schema, which avoids restart client
                 // after users changed the compatibility strategy to make the schema is compatible.
                 tryRegisterSchema(cnx, loopEndDueToSchemaRegisterNeeded.msg, loopEndDueToSchemaRegisterNeeded.callback,
                     expectedEpoch);
-            } else if (!failedIncompatibleSchema && compareAndSetState(State.RegisteringSchema, State.Ready)) {
-                // "Event 2-1" or "Event 3-2".
+            } else if (pausedSendingToPreservePublishOrderOnSchemaRegFailure) {
+                // Nothing to do if the event is "Event 3-1-1", just keep stuck.
+                return;
+            } else if (compareAndSetState(State.RegisteringSchema, State.Ready)) {
+                // "Event 2-1" or "Event 3-1-2" or "Event 3-2".
                 // "pendingMessages" has more messages to register new schema.
                 // This operation will not be conflict with another schema registration because both operations are
                 // attempt to acquire the same lock "ProducerImpl.this".
                 tryRegisterSchema(cnx, loopEndDueToSchemaRegisterNeeded.msg, loopEndDueToSchemaRegisterNeeded.callback,
                         expectedEpoch);
             }
-            // Nothing to do if the event is "Event 3-1-1", just keep stuck.
+            // Schema registration will trigger a new "recoverProcessOpSendMsgFrom", so return here. If failed to switch
+            // state, it means another task will trigger a new "recoverProcessOpSendMsgFrom".
             return;
         } else if (latestMsgAttemptedRegisteredSchema != null) {
             // Event 2-2 or "Event 3-1-2".