[branch-2.10][test]Run and fix tests

buildtools/src/main/java/org/apache/pulsar/tests/MockitoCleanupListener.java

@@ -39,7 +39,8 @@ public class MockitoCleanupListener extends BetweenTestClassesListenerAdapter {
     protected void onBetweenTestClasses(Class<?> endedTestClass, Class<?> startedTestClass) {
         if (MOCKITO_CLEANUP_ENABLED) {
             if (MockitoThreadLocalStateCleaner.INSTANCE.isEnabled()) {
-                LOG.info("Cleaning up Mockito's ThreadSafeMockingProgress.MOCKING_PROGRESS_PROVIDER thread local state.");
+                LOG.info("Cleaning up Mockito's ThreadSafeMockingProgress.MOCKING_PROGRESS_PROVIDER "
+                        + "thread local state.");
                 MockitoThreadLocalStateCleaner.INSTANCE.cleanup();
             }
             cleanupMockitoInline();

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtil.java

@@ -60,6 +60,13 @@ public static long[] andAckSet(long[] firstAckSet, long[] secondAckSet) {
         return ackSet;
     }
 
+    public static boolean isAckSetEmpty(long[] ackSet) {
+        BitSetRecyclable bitSet =  BitSetRecyclable.create().resetWords(ackSet);
+        boolean isEmpty = bitSet.isEmpty();
+        bitSet.recycle();
+        return isEmpty;
+    }
+
     //This method is compare two position which position is bigger than another one.
     //When the ledgerId and entryId in this position is same to another one and two position all have ack set, it will
     //compare the ack set next bit index is bigger than another one.

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

@@ -5068,7 +5068,8 @@ private void internalGetReplicatedSubscriptionStatusForNonPartitionedTopic(Async
             // Redirect the request to the appropriate broker if this broker is not the owner of the topic
             validateTopicOwnership(topicName, authoritative);
 
-            getReplicatedSubscriptionStatusFromLocalBroker(topicName, subName).get();
+            Map res = getReplicatedSubscriptionStatusFromLocalBroker(topicName, subName).get();
+            asyncResponse.resume(res);
         } catch (Exception e) {
             log.error("[{}] Failed to get replicated subscription status on {} {}", clientAppId(),
                     topicName, subName, e);

pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java

@@ -64,8 +64,8 @@
 import org.apache.pulsar.broker.resources.ClusterResources;
 import org.apache.pulsar.broker.stats.prometheus.metrics.Summary;
 import org.apache.pulsar.client.admin.PulsarAdminException;
-import org.apache.pulsar.common.naming.NamespaceBundle;
 import org.apache.pulsar.client.util.ExecutorProvider;
+import org.apache.pulsar.common.naming.NamespaceBundle;
 import org.apache.pulsar.common.naming.NamespaceBundleFactory;
 import org.apache.pulsar.common.naming.NamespaceName;
 import org.apache.pulsar.common.naming.ServiceUnitId;

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java

@@ -20,6 +20,7 @@
 package org.apache.pulsar.broker.service;
 
 import static org.apache.bookkeeper.mledger.util.PositionAckSetUtil.andAckSet;
+import static org.apache.bookkeeper.mledger.util.PositionAckSetUtil.isAckSetEmpty;
 import com.google.common.collect.ImmutableList;
 import io.netty.buffer.ByteBuf;
 import java.util.ArrayList;
@@ -246,6 +247,18 @@ public int filterEntriesForConsumer(@Nullable EntryWrapper[] entryWrapper, int e
                                 // if actSet is null, use pendingAck ackSet
                                 ackSet = positionInPendingAck.getAckSet();
                             }
+                            // if the result of pendingAckSet(in pendingAckHandle) AND the ackSet(in cursor) is empty
+                            // filter this entry
+                            if (isAckSetEmpty(ackSet)) {
+                                entries.set(i, null);
+                                entry.release();
+                                continue;
+                            }
+                        } else {
+                            // filter non-batch message in pendingAck state
+                            entries.set(i, null);
+                            entry.release();
+                            continue;
                         }
                     }
                 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java

@@ -21,10 +21,10 @@
 import static com.google.common.base.Preconditions.checkArgument;
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.apache.commons.lang3.StringUtils.isBlank;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.github.benmanes.caffeine.cache.CacheLoader;
 import com.github.benmanes.caffeine.cache.Caffeine;
 import com.github.benmanes.caffeine.cache.LoadingCache;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.BoundType;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Range;

pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceOwnershipListenerTests.java

@@ -19,6 +19,7 @@
 package org.apache.pulsar.broker.namespace;
 
 import com.google.common.collect.Sets;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.pulsar.broker.service.BrokerTestBase;
 import org.apache.pulsar.client.admin.PulsarAdminException;
@@ -30,11 +31,9 @@
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
-
 import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.testng.Assert.assertTrue;
@@ -55,7 +54,7 @@ protected void cleanup() throws Exception {
     }
 
     @Test
-    public void testNamespaceBundleOwnershipListener() throws PulsarAdminException, InterruptedException, PulsarClientException {
+    public void testNamespaceBundleOwnershipListener() throws Exception {
 
         final CountDownLatch countDownLatch = new CountDownLatch(2);
         final AtomicBoolean onLoad = new AtomicBoolean(false);

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

@@ -171,6 +171,84 @@ private void testIndividualAckAbortFilterAckSetInPendingAckState() throws Except
         assertNull(consumer.receive(2, TimeUnit.SECONDS));
     }
 
+
+    @Test(dataProvider="enableBatch")
+    private void testFilterMsgsInPendingAckStateWhenConsumerDisconnect(boolean enableBatch) throws Exception {
+        final String topicName = NAMESPACE1 + "/testFilterMsgsInPendingAckStateWhenConsumerDisconnect-" + enableBatch;
+        final int count = 10;
+
+        @Cleanup
+        Producer<Integer> producer = null;
+        if (enableBatch) {
+            producer = pulsarClient
+                    .newProducer(Schema.INT32)
+                    .topic(topicName)
+                    .enableBatching(true)
+                    .batchingMaxPublishDelay(1, TimeUnit.HOURS)
+                    .batchingMaxMessages(count).create();
+        } else {
+            producer = pulsarClient
+                    .newProducer(Schema.INT32)
+                    .topic(topicName)
+                    .enableBatching(false).create();
+        }
+
+        @Cleanup
+        Consumer<Integer> consumer = pulsarClient
+                .newConsumer(Schema.INT32)
+                .topic(topicName)
+                .isAckReceiptEnabled(true)
+                .subscriptionName("test")
+                .subscriptionType(SubscriptionType.Shared)
+                .enableBatchIndexAcknowledgment(true)
+                .subscribe();
+
+        for (int i = 0; i < count; i++) {
+            producer.sendAsync(i);
+        }
+
+        Transaction txn1 = getTxn();
+
+        Transaction txn2 = getTxn();
+
+
+        // txn1 ack half of messages and don't end the txn1
+        for (int i = 0; i < count / 2; i++) {
+            consumer.acknowledgeAsync(consumer.receive().getMessageId(), txn1).get();
+        }
+
+        // txn2 ack the rest half of messages and commit tnx2
+        for (int i = count / 2; i < count; i++) {
+            consumer.acknowledgeAsync(consumer.receive().getMessageId(), txn2).get();
+        }
+        // commit txn2
+        txn2.commit().get();
+
+        // close and re-create consumer
+        consumer.close();
+        consumer = pulsarClient
+                .newConsumer(Schema.INT32)
+                .topic(topicName)
+                .isAckReceiptEnabled(true)
+                .subscriptionName("test")
+                .subscriptionType(SubscriptionType.Shared)
+                .enableBatchIndexAcknowledgment(true)
+                .subscribe();
+
+        Message<Integer> message = consumer.receive(3, TimeUnit.SECONDS);
+        Assert.assertNull(message);
+
+        // abort txn1
+        txn1.abort().get();
+        // after txn1 aborted, consumer will receive messages txn1 contains
+        int receiveCounter = 0;
+        while((message = consumer.receive(3, TimeUnit.SECONDS)) != null) {
+            Assert.assertEquals(message.getValue().intValue(), receiveCounter);
+            receiveCounter ++;
+        }
+        Assert.assertEquals(receiveCounter, count / 2);
+    }
+
     @Test(dataProvider="enableBatch")
     private void produceCommitTest(boolean enableBatch) throws Exception {
         @Cleanup

pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImplTest.java

@@ -34,8 +34,6 @@
 import org.apache.pulsar.client.api.EncodedAuthenticationParameterSupport;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.impl.auth.AuthenticationDisabled;
-import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
-import org.testng.Assert;
 import org.testng.annotations.Test;
 
 /**

pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/BitSetRecyclableRecyclableTest.java

@@ -45,4 +45,22 @@ public void testResetWords() {
         Assert.assertTrue(bitset1.get(128));
         Assert.assertFalse(bitset1.get(256));
     }
+
+    @Test
+    public void testBitSetEmpty() {
+        BitSetRecyclable bitSet = BitSetRecyclable.create();
+        bitSet.set(0, 5);
+        bitSet.clear(1);
+        bitSet.clear(2);
+        bitSet.clear(3);
+        long[] array = bitSet.toLongArray();
+        Assert.assertFalse(bitSet.isEmpty());
+        Assert.assertFalse(BitSetRecyclable.create().resetWords(array).isEmpty());
+        bitSet.clear(0);
+        bitSet.clear(4);
+        Assert.assertTrue(bitSet.isEmpty());
+        long[] array1 = bitSet.toLongArray();
+        Assert.assertTrue(BitSetRecyclable.create().resetWords(array1).isEmpty());
+        bitSet.recycle();
+    }
 }

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceCache.java

@@ -18,7 +18,6 @@
  */
 package org.apache.pulsar.functions.instance;
 
-import io.netty.util.concurrent.DefaultThreadFactory;
 import lombok.Getter;
 
 import java.util.concurrent.Executors;

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/ClusterServiceCoordinator.java

@@ -31,9 +31,6 @@
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.client.util.ExecutorProvider;
-
-import java.util.HashMap;
-import java.util.Map;
 import java.util.function.Supplier;
 
 @Slf4j