[Transaction] Fix the Github CI Test (apache#8541)

Motivation
Try to fix the Github CI.

## implement
because we consume all transaction messages than have been produced. but commit marker may not delete, because when a commit marker operation is processing, when we ack the position will not allow to do delete operation of commit marker, so the commit marker may not delete. we should handle this case.

Author: congbobo184

pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckInMemoryDeleteTest.java

@@ -95,7 +95,7 @@ protected void cleanup() {
     }
 
     @Test
-    private void txnAckTestNoBatchAndSharedSubMemoryDeleteTest() throws Exception {
+    public void txnAckTestNoBatchAndSharedSubMemoryDeleteTest() throws Exception {
         String normalTopic = NAMESPACE1 + "/normal-topic";
         String subscriptionName = "test";
 
@@ -178,7 +178,7 @@ private void txnAckTestNoBatchAndSharedSubMemoryDeleteTest() throws Exception {
     }
 
     @Test
-    private void txnAckTestBatchAndSharedSubMemoryDeleteTest() throws Exception {
+    public void txnAckTestBatchAndSharedSubMemoryDeleteTest() throws Exception {
         String normalTopic = NAMESPACE1 + "/normal-topic";
         String subscriptionName = "test";
 

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

@@ -19,17 +19,29 @@
 package org.apache.pulsar.client.impl;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.testng.Assert.assertTrue;
 import static org.testng.Assert.fail;
 
 import com.google.common.collect.Sets;
 
+import java.lang.reflect.Field;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
+
 import lombok.Cleanup;
 import lombok.extern.slf4j.Slf4j;
+
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.service.Topic;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
 import org.apache.pulsar.broker.transaction.TransactionTestBase;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.Message;
@@ -51,6 +63,7 @@
 import org.apache.pulsar.common.policies.data.ClusterData;
 import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
 import org.apache.pulsar.common.policies.data.TenantInfo;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
 import org.testng.Assert;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
@@ -317,7 +330,7 @@ private void txnAckTest(boolean batchEnable, int maxBatchSize,
 
     @Test
     public void txnMessageAckTest() throws Exception {
-        final String topic = TOPIC_MESSAGE_ACK_TEST;
+        String topic = TOPIC_MESSAGE_ACK_TEST;
         final String subName = "test";
         @Cleanup
         Consumer<byte[]> consumer = pulsarClient
@@ -382,8 +395,46 @@ public void txnMessageAckTest() throws Exception {
 
         message = consumer.receive(2, TimeUnit.SECONDS);
         Assert.assertNull(message);
-
-        markDeletePositionCheck(topic, subName, true);
+        for (int partition = 0; partition < TOPIC_PARTITION; partition ++) {
+            topic = TopicName.get(topic).getPartition(partition).toString();
+            boolean exist = false;
+            for (int i = 0; i < getPulsarServiceList().size(); i++) {
+
+                Field field = BrokerService.class.getDeclaredField("topics");
+                field.setAccessible(true);
+                ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
+                        (ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>>) field
+                                .get(getPulsarServiceList().get(i).getBrokerService());
+                CompletableFuture<Optional<Topic>> topicFuture = topics.get(topic);
+
+                if (topicFuture != null) {
+                    Optional<Topic> topicOptional = topicFuture.get();
+                    if (topicOptional.isPresent()) {
+                        PersistentSubscription persistentSubscription =
+                                (PersistentSubscription) topicOptional.get().getSubscription(subName);
+                        Position markDeletePosition = persistentSubscription.getCursor().getMarkDeletedPosition();
+                        Position lastConfirmedEntry = persistentSubscription.getCursor()
+                                .getManagedLedger().getLastConfirmedEntry();
+                        exist = true;
+                        if (!markDeletePosition.equals(lastConfirmedEntry)) {
+                            //this because of the transaction commit marker have't delete
+                            //delete commit marker after ack position
+                            //when delete commit marker operation is processing, next delete operation will not do again
+                            //when delete commit marker operation finish, it can run next delete commit marker operation
+                            //so this test may not delete all the position in this manageLedger.
+                            Position markerPosition = ((ManagedLedgerImpl) persistentSubscription.getCursor()
+                                    .getManagedLedger()).getNextValidPosition((PositionImpl) markDeletePosition);
+                            //marker is the lastConfirmedEntry, after commit the marker will only be write in
+                            if (!markerPosition.equals(lastConfirmedEntry)) {
+                                log.error("Mark delete position is not commit marker position!");
+                                fail();
+                            }
+                        }
+                    }
+                }
+            }
+            assertTrue(exist);
+        }
 
         log.info("receive transaction messages count: {}", receiveCnt);
     }