[improve] [proxy] Not close the socket if lookup failed caused by too…

… many requests (#21216)

Motivation: The Pulsar client will close the socket if it receives a `ServiceNotReady` error when doing a lookup. The Broker will respond to the client with a `TooManyRequests` error if there are too many lookup requests in progress, but the Pulsar Proxy responds to the client with a `ServiceNotReady` error in the same scenario.

Modifications: Make Pulsar Proxy respond to the client with a `TooManyRequests` error if there are too many lookup requests in progress.
(cherry picked from commit d6c3fa4)

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java

@@ -113,7 +113,7 @@ public void handleLookup(CommandLookupTopic lookup) {
                 log.debug("Lookup Request ID {} from {} rejected - {}.", clientRequestId, clientAddress,
                         throttlingErrorMessage);
             }
-            proxyConnection.ctx().writeAndFlush(Commands.newLookupErrorResponse(ServerError.ServiceNotReady,
+            proxyConnection.ctx().writeAndFlush(Commands.newLookupErrorResponse(ServerError.TooManyRequests,
                     throttlingErrorMessage, clientRequestId));
         }
 

pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyLookupThrottlingTest.java

@@ -20,19 +20,29 @@
 
 import static org.mockito.Mockito.doReturn;
 import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
 
 import java.util.Optional;
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 
 import lombok.Cleanup;
 
+import org.apache.pulsar.broker.BrokerTestUtil;
 import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
 import org.apache.pulsar.broker.authentication.AuthenticationService;
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.BinaryProtoLookupService;
+import org.apache.pulsar.client.impl.ClientCnx;
+import org.apache.pulsar.client.impl.LookupService;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.ServiceNameResolver;
 import org.apache.pulsar.common.configuration.PulsarConfigurationLoader;
+import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.metadata.impl.ZKMetadataStore;
+import org.awaitility.reflect.WhiteboxImpl;
 import org.mockito.Mockito;
 import org.testng.Assert;
 import org.testng.annotations.AfterMethod;
@@ -112,4 +122,34 @@ public void testLookup() throws Exception {
 
         Assert.assertEquals(LookupProxyHandler.REJECTED_PARTITIONS_METADATA_REQUESTS.get(), 5.0d);
     }
+
+    @Test
+    public void testLookupThrottling() throws Exception {
+        PulsarClientImpl client = (PulsarClientImpl) PulsarClient.builder()
+                .serviceUrl(proxyService.getServiceUrl()).build();
+        String tpName = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+        LookupService lookupService = client.getLookup();
+        assertTrue(lookupService instanceof BinaryProtoLookupService);
+        ServiceNameResolver serviceNameResolver =
+                WhiteboxImpl.getInternalState(lookupService, "serviceNameResolver");
+        ClientCnx lookupConnection = client.getCnxPool().getConnection(serviceNameResolver.resolveHost()).join();
+
+        // Make no permits to lookup.
+        Semaphore lookupSemaphore = proxyService.getLookupRequestSemaphore();
+        int availablePermits = lookupSemaphore.availablePermits();
+        lookupSemaphore.acquire(availablePermits);
+
+        // Verify will receive too many request exception, and the socket will not be closed.
+        try {
+            lookupService.getBroker(TopicName.get(tpName)).get();
+            fail("Expected too many request error.");
+        } catch (Exception ex) {
+            assertTrue(ex.getMessage().contains("Too many"));
+        }
+        assertTrue(lookupConnection.ctx().channel().isActive());
+
+        // cleanup.
+        lookupSemaphore.release(availablePermits);
+        client.close();
+    }
 }