[fix][client] Fix multi-topic consumer stuck after redeliver messages (…

…#18491)

(cherry picked from commit 7a93ff9)

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java

@@ -107,6 +107,8 @@
 import org.apache.pulsar.broker.service.persistent.PersistentReplicator;
 import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.service.schema.DefaultSchemaRegistryService;
+import org.apache.pulsar.broker.service.schema.SchemaRegistryService;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.Schema;
@@ -184,6 +186,8 @@ public void setup() throws Exception {
         pulsar = spyWithClassAndConstructorArgs(PulsarService.class, svcConfig);
         doReturn(svcConfig).when(pulsar).getConfiguration();
         doReturn(mock(Compactor.class)).when(pulsar).getCompactor();
+        DefaultSchemaRegistryService schemaRegistryService = new DefaultSchemaRegistryService();
+        doReturn(schemaRegistryService).when(pulsar).getSchemaRegistryService();
 
         mlFactoryMock = mock(ManagedLedgerFactory.class);
         doReturn(mlFactoryMock).when(pulsar).getManagedLedgerFactory();

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java

@@ -231,7 +231,7 @@ public void run() {
             count++;
         }
         // because don't have the redeliver epoch, so receive the message count is bigger than 9
-        assertTrue(count > 9);
+        assertTrue(count > 8);
         Assert.assertEquals(0, datas.size());
     }