[fix][flask-test] Fix and improve LookupRetryTest (#17848)

Fixes #17785

### Motivation

The `failureMap` need to be clear after run per unit test.

### Modifications

Clear `failureMap` after run per unit test, and only run once `setup()`/`cleanup()` to reduce execution time.

### Matching PR in forked repository

PR in forked repository: coderzc#6

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

@@ -20,14 +20,12 @@
 
 import static org.apache.pulsar.common.protocol.Commands.newLookupErrorResponse;
 import static org.apache.pulsar.common.protocol.Commands.newPartitionMetadataResponse;
-
 import com.google.common.collect.Sets;
 import java.util.Queue;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
-
 import org.apache.pulsar.broker.PulsarService;
 import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
@@ -44,20 +42,18 @@
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.policies.data.ClusterData;
 import org.apache.pulsar.common.policies.data.TenantInfoImpl;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.testng.Assert;
-import org.testng.annotations.AfterMethod;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
 public class LookupRetryTest extends MockedPulsarServiceBaseTest {
-    private static final Logger log = LoggerFactory.getLogger(LookupRetryTest.class);
     private static final String subscription = "reader-sub";
     private final AtomicInteger connectionsCreated = new AtomicInteger(0);
     private final ConcurrentHashMap<String, Queue<LookupError>> failureMap = new ConcurrentHashMap<>();
 
-    @BeforeMethod
+    @BeforeClass
     @Override
     protected void setup() throws Exception {
         conf.setTopicLevelPoliciesEnabled(false);
@@ -69,8 +65,6 @@ protected void setup() throws Exception {
         admin.tenants().createTenant("public",
                 new TenantInfoImpl(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
         admin.namespaces().createNamespace("public/default", Sets.newHashSet("test"));
-
-        connectionsCreated.set(0);
     }
 
     @Override
@@ -94,12 +88,18 @@ protected ServerCnx newServerCnx(PulsarService pulsar, String listenerName) thro
         };
     }
 
-    @AfterMethod(alwaysRun = true)
+    @AfterClass(alwaysRun = true)
     @Override
     protected void cleanup() throws Exception {
         super.internalCleanup();
     }
 
+    @BeforeMethod(alwaysRun = true)
+    public void reset() {
+        connectionsCreated.set(0);
+        failureMap.clear();
+    }
+
     PulsarClient newClient() throws Exception {
         return PulsarClient.builder()
             .serviceUrl(pulsar.getBrokerServiceUrl())
@@ -244,7 +244,6 @@ public void testCloseConnectionOnBrokerTimeout() throws Exception {
 
         try (PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(lookupUrl)
                 .maxNumberOfRejectedRequestPerConnection(100)
-                .maxNumberOfRejectedRequestPerConnection(1)
                 .connectionTimeout(2, TimeUnit.SECONDS)
                 .operationTimeout(1, TimeUnit.SECONDS)
                 .lookupTimeout(10, TimeUnit.SECONDS)