[fix][test] Stabilize SequenceIdWithErrorTest by fencing after first publish to avoid empty-ledger deletion and send timeout (apache#24861)

(cherry picked from commit b0ce575)
(cherry picked from commit fe84bdb)

Author: Denovo1998

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

@@ -22,6 +22,7 @@
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import java.util.Collections;
+import java.util.concurrent.TimeUnit;
 import lombok.Cleanup;
 import org.apache.bookkeeper.mledger.ManagedLedger;
 import org.apache.bookkeeper.mledger.ManagedLedgerFactory;
@@ -55,6 +56,13 @@ public void testCheckSequenceId() throws Exception {
         Consumer<String> consumer = client.newConsumer(Schema.STRING).topic(topicName).subscriptionName("sub")
                 .subscribe();
 
+        // Create a producer
+        Producer<String> producer = client.newProducer(Schema.STRING).topic(topicName).create();
+        // Move the fence timing to after the first message is successfully written
+        // The current ledger is not empty, the Broker recovery will not take the abnormal path of
+        // "deleting empty ledger + unable to find old ledger"
+        producer.send("Hello-0");
+
         // Fence the topic by opening the ManagedLedger for the topic outside the Pulsar broker. This will cause the
         // broker to fail subsequent send operation and it will trigger a recover
         EventLoopGroup eventLoopGroup = new NioEventLoopGroup(1);
@@ -66,15 +74,12 @@ public void testCheckSequenceId() throws Exception {
         ml.close();
         clientFactory.close();
 
-        // Create a producer
-        Producer<String> producer = client.newProducer(Schema.STRING).topic(topicName).create();
-
-        for (int i = 0; i < N; i++) {
+        for (int i = 1; i < N; i++) {
             producer.send("Hello-" + i);
         }
 
         for (int i = 0; i < N; i++) {
-            Message<String> msg = consumer.receive();
+            Message<String> msg = consumer.receive(10, TimeUnit.SECONDS);
             assertEquals(msg.getValue(), "Hello-" + i);
             assertEquals(msg.getSequenceId(), i);
             consumer.acknowledge(msg);