[fix][proxy] Fix incorrect client error when calling get topic metadata

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java

@@ -1382,6 +1382,47 @@ public static PulsarClientException getPulsarClientException(ServerError error,
         }
     }
 
+    public static ServerError revertClientExToErrorCode(PulsarClientException ex) {
+        if (ex instanceof PulsarClientException.AuthenticationException) {
+            return ServerError.AuthenticationError;
+        } else if (ex instanceof PulsarClientException.AuthorizationException) {
+            return ServerError.AuthorizationError;
+        } else if (ex instanceof PulsarClientException.ProducerBusyException) {
+            return ServerError.ProducerBusy;
+        } else if (ex instanceof PulsarClientException.ConsumerBusyException) {
+            return ServerError.ConsumerBusy;
+        } else if (ex instanceof PulsarClientException.BrokerMetadataException) {
+            return ServerError.MetadataError;
+        } else if (ex instanceof PulsarClientException.BrokerPersistenceException) {
+            return ServerError.PersistenceError;
+        } else if (ex instanceof PulsarClientException.TooManyRequestsException) {
+            return ServerError.TooManyRequests;
+        } else if (ex instanceof PulsarClientException.LookupException) {
+            return ServerError.ServiceNotReady;
+        } else if (ex instanceof PulsarClientException.ProducerBlockedQuotaExceededError) {
+            return ServerError.ProducerBlockedQuotaExceededError;
+        } else if (ex instanceof PulsarClientException.ProducerBlockedQuotaExceededException) {
+            return ServerError.ProducerBlockedQuotaExceededException;
+        } else if (ex instanceof PulsarClientException.TopicTerminatedException) {
+            return ServerError.TopicTerminatedError;
+        } else if (ex instanceof PulsarClientException.IncompatibleSchemaException) {
+            return ServerError.IncompatibleSchema;
+        } else if (ex instanceof PulsarClientException.TopicDoesNotExistException) {
+            return ServerError.TopicNotFound;
+        } else if (ex instanceof PulsarClientException.SubscriptionNotFoundException) {
+            return ServerError.SubscriptionNotFound;
+        } else if (ex instanceof PulsarClientException.ConsumerAssignException) {
+            return ServerError.ConsumerAssignError;
+        } else if (ex instanceof PulsarClientException.NotAllowedException) {
+            return ServerError.NotAllowedError;
+        } else if (ex instanceof PulsarClientException.TransactionConflictException) {
+            return ServerError.TransactionConflict;
+        } else if (ex instanceof PulsarClientException.ProducerFencedException) {
+            return ServerError.ProducerFenced;
+        }
+        return ServerError.UnknownError;
+    }
+
     public void close() {
        if (ctx != null) {
            ctx.close();

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

@@ -247,8 +247,9 @@ private void handlePartitionMetadataResponse(CommandPartitionedTopicMetadata par
                 if (t != null) {
                     log.warn("[{}] failed to get Partitioned metadata : {}", topicName.toString(),
                         t.getMessage(), t);
-                    writeAndFlush(Commands.newLookupErrorResponse(getServerError(t),
-                        t.getMessage(), clientRequestId));
+                    PulsarClientException pce = PulsarClientException.unwrap(t);
+                    writeAndFlush(Commands.newLookupErrorResponse(clientCnx.revertClientExToErrorCode(pce),
+                            t.getMessage(), clientRequestId));
                 } else {
                     writeAndFlush(
                         Commands.newPartitionMetadataResponse(r.partitions, clientRequestId));

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

@@ -24,6 +24,7 @@
 import static org.mockito.Mockito.doReturn;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.EventLoopGroup;
 import io.netty.util.concurrent.DefaultThreadFactory;
@@ -52,6 +53,7 @@
 import org.apache.pulsar.client.api.MessageRoutingMode;
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.SubscriptionInitialPosition;
 import org.apache.pulsar.client.api.SubscriptionType;
@@ -68,13 +70,15 @@
 import org.apache.pulsar.common.api.proto.ProtocolVersion;
 import org.apache.pulsar.common.configuration.PulsarConfigurationLoader;
 import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
 import org.apache.pulsar.common.policies.data.ClusterData;
 import org.apache.pulsar.common.policies.data.RetentionPolicies;
 import org.apache.pulsar.common.policies.data.TenantInfo;
 import org.apache.pulsar.common.policies.data.TenantInfoImpl;
 import org.apache.pulsar.common.policies.data.TopicType;
 import org.apache.pulsar.common.protocol.Commands;
 import org.apache.pulsar.common.schema.SchemaInfo;
+import org.apache.pulsar.common.util.FutureUtil;
 import org.apache.pulsar.common.util.netty.EventLoopUtil;
 import org.apache.pulsar.metadata.impl.ZKMetadataStore;
 import org.mockito.Mockito;
@@ -140,6 +144,13 @@ protected void initializeProxyConfig() throws Exception {
         proxyClientAuthentication.start();
     }
 
+    @Override
+    protected void doInitConf() throws Exception {
+        super.doInitConf();
+        conf.setAllowAutoTopicCreationType(TopicType.PARTITIONED);
+        conf.setDefaultNumPartitions(1);
+    }
+
     @Override
     @AfterClass(alwaysRun = true)
     protected void cleanup() throws Exception {
@@ -415,6 +426,35 @@ public void testProtocolVersionAdvertisement() throws Exception {
         }
     }
 
+    @Test
+    public void testGetPartitionedMetadataErrorCode() throws Exception {
+        final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+        // Trigger partitioned metadata creation.
+        PulsarClientImpl brokerClient = (PulsarClientImpl) pulsarClient;
+        PartitionedTopicMetadata brokerMetadata =
+                brokerClient.getPartitionedTopicMetadata(topic, true, true).get();
+        assertEquals(brokerMetadata.partitions, 1);
+        assertEquals(pulsar.getPulsarResources().getNamespaceResources().getPartitionedTopicResources()
+                .getPartitionedTopicMetadataAsync(TopicName.get(topic)).get().get().partitions, 1);
+        // Verify: Proxy never rewrite error code.
+        ClientConfigurationData proxyClientConf = new ClientConfigurationData();
+        proxyClientConf.setServiceUrl(proxyService.getServiceUrl());
+        PulsarClientImpl proxyClient =
+                (PulsarClientImpl) getClientActiveConsumerChangeNotSupported(proxyClientConf);
+        PartitionedTopicMetadata proxyMetadata =
+                proxyClient.getPartitionedTopicMetadata(topic, false, false).get();
+        assertEquals(proxyMetadata.partitions, 1);
+        try {
+            proxyClient.getPartitionedTopicMetadata(topic + "-partition-0", false, false).get();
+        } catch (Exception ex) {
+            assertTrue(FutureUtil.unwrapCompletionException(ex)
+                    instanceof PulsarClientException.TopicDoesNotExistException);
+        }
+        // cleanup.
+        proxyClient.close();
+        admin.topics().deletePartitionedTopic(topic);
+    }
+
     @Test
     public void testGetClientVersion() throws Exception {
         @Cleanup