[fix][broker] Flaky-test: TopicTransactionBufferTest.testMessagePublishInOrder (apache#24826)

Signed-off-by: Dream95 <zhou_8621@163.com>
Co-authored-by: Lari Hotari <lhotari@users.noreply.github.com>
(cherry picked from commit 8828734)
(cherry picked from commit 74c366c)

Author: 2 people

pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java

@@ -229,6 +229,11 @@ public CompletableFuture<TransactionMeta> getTransactionMeta(TxnID txnID) {
         return CompletableFuture.completedFuture(null);
     }
 
+    @VisibleForTesting
+    public void setPublishFuture(CompletableFuture<Position> publishFuture) {
+        this.publishFuture = publishFuture;
+    }
+
     @VisibleForTesting
     public CompletableFuture<Position> getPublishFuture() {
         return publishFuture;
@@ -295,7 +300,7 @@ public CompletableFuture<Position> appendBufferToTxn(TxnID txnId, long sequenceI
                         "Transaction Buffer recover failed, the current state is: " + getState()));
             }
         }).whenComplete(((position, throwable) -> buffer.release()));
-        publishFuture = future;
+        setPublishFuture(future);
         return future;
     }
 

pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/utils/TransactionBufferTestImpl.java

@@ -18,36 +18,20 @@
  */
 package org.apache.pulsar.broker.transaction.buffer.utils;
 
-import java.util.concurrent.CompletableFuture;
 import lombok.Setter;
-import org.apache.bookkeeper.mledger.Position;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
 import org.apache.pulsar.broker.transaction.buffer.impl.TopicTransactionBuffer;
 
 public class TransactionBufferTestImpl extends TopicTransactionBuffer {
-    @Setter
-    public CompletableFuture<Void> transactionBufferFuture = null;
     @Setter
     public State state = null;
-    @Setter
-    public CompletableFuture<Position> publishFuture = null;
 
     public TransactionBufferTestImpl(PersistentTopic topic) {
         super(topic);
     }
 
-    @Override
-    public CompletableFuture<Void> getTransactionBufferFuture() {
-        return transactionBufferFuture == null ? super.getTransactionBufferFuture() : transactionBufferFuture;
-    }
-
     @Override
     public State getState() {
         return state == null ? super.getState() : state;
     }
-
-    @Override
-    public CompletableFuture<Position> getPublishFuture() {
-       return publishFuture == null ? super.getPublishFuture() : publishFuture;
-    }
 }