@Override
    protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata partitionMetadata) {
        checkArgument(state == State.Connected);
        final long requestId = partitionMetadata.getRequestId();
        if (log.isDebugEnabled()) {
            log.debug("[{}] Received PartitionMetadataLookup from {} for {}", partitionMetadata.getTopic(),
                    remoteAddress, requestId);
        }

        TopicName topicName = validateTopicName(partitionMetadata.getTopic(), requestId, partitionMetadata);
        if (topicName == null) {
            return;
        }

        if (!this.service.getPulsar().isRunning()) {
            if (log.isDebugEnabled()) {
                log.debug("[{}] Failed PartitionMetadataLookup from {} for {} "
                                + "due to pulsar service is not ready: {} state",
                        partitionMetadata.getTopic(), remoteAddress, requestId,
                        this.service.getPulsar().getState().toString());
            }
            writeAndFlush(Commands.newPartitionMetadataResponse(ServerError.ServiceNotReady,
                    "Failed due to pulsar service is not ready", requestId));
            return;
        }

        final Semaphore lookupSemaphore = service.getLookupRequestSemaphore();
        if (lookupSemaphore.tryAcquire()) {
            isTopicOperationAllowed(topicName, TopicOperation.LOOKUP, authenticationData, originalAuthData).thenApply(
                    isAuthorized -> {
                if (isAuthorized) {
                    // Get if exists, respond not found error if not exists.
                    getBrokerService().isAllowAutoTopicCreationAsync(topicName).thenAccept(brokerAllowAutoCreate -> {
                        boolean autoCreateIfNotExist = partitionMetadata.isMetadataAutoCreationEnabled()
                                && brokerAllowAutoCreate;
                        if (!autoCreateIfNotExist) {
                            NamespaceService namespaceService = getBrokerService().getPulsar().getNamespaceService();
                            namespaceService.checkTopicExists(topicName).thenAccept(topicExistsInfo -> {
                                lookupSemaphore.release();
                                if (!topicExistsInfo.isExists()) {
                                    writeAndFlush(Commands.newPartitionMetadataResponse(
                                            ServerError.TopicNotFound, "", requestId));
                                } else if (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)) {
                                    commandSender.sendPartitionMetadataResponse(topicExistsInfo.getPartitions(),
                                            requestId);
                                } else {
                                    commandSender.sendPartitionMetadataResponse(0, requestId);
                                }
                                // release resources.
                                topicExistsInfo.recycle();
                            }).exceptionally(ex -> {
                                lookupSemaphore.release();
                                log.error("{} {} Failed to get partition metadata", topicName,
                                        ServerCnx.this.toString(), ex);
                                writeAndFlush(
                                        Commands.newPartitionMetadataResponse(ServerError.MetadataError,
                                                "Failed to get partition metadata",
                                                requestId));
                                return null;
                            });
                        } else {
                            // Get if exists, create a new one if not exists.
                            unsafeGetPartitionedTopicMetadataAsync(getBrokerService().pulsar(), topicName)
                                .whenComplete((metadata, ex) -> {
                                    lookupSemaphore.release();
                                    if (ex == null) {
                                        int partitions = metadata.partitions;
                                        commandSender.sendPartitionMetadataResponse(partitions, requestId);
                                    } else {
                                        if (ex instanceof PulsarClientException) {
                                            log.warn("Failed to authorize {} at [{}] on topic {} : {}", getRole(),
                                                    remoteAddress, topicName, ex.getMessage());
                                            commandSender.sendPartitionMetadataResponse(ServerError.AuthorizationError,
                                                    ex.getMessage(), requestId);
                                        } else {
                                            log.warn("Failed to get Partitioned Metadata [{}] {}: {}", remoteAddress,
                                                    topicName, ex.getMessage(), ex);
                                            ServerError error = ServerError.ServiceNotReady;
                                            if (ex instanceof MetadataStoreException) {
                                                error = ServerError.MetadataError;
                                            } else if (ex instanceof RestException restException){
                                                int responseCode = restException.getResponse().getStatus();
                                                if (responseCode == NOT_FOUND.getStatusCode()){
                                                    error = ServerError.TopicNotFound;
                                                } else if (responseCode < INTERNAL_SERVER_ERROR.getStatusCode()){
                                                    error = ServerError.MetadataError;
                                                }
                                            }
                                            commandSender.sendPartitionMetadataResponse(error, ex.getMessage(),
                                                    requestId);
                                        }
                                    }
                                });
                        }
                    });
                } else {
                    final String msg = "Client is not authorized to Get Partition Metadata";
                    log.warn("[{}] {} with role {} on topic {}", remoteAddress, msg, getPrincipal(), topicName);
                    writeAndFlush(
                            Commands.newPartitionMetadataResponse(ServerError.AuthorizationError, msg, requestId));
                    lookupSemaphore.release();
                }
                return null;
            }).exceptionally(ex -> {
                logAuthException(remoteAddress, "partition-metadata", getPrincipal(), Optional.of(topicName), ex);
                final String msg = "Exception occurred while trying to authorize get Partition Metadata";
                writeAndFlush(Commands.newPartitionMetadataResponse(ServerError.AuthorizationError, msg,
                        requestId));
                lookupSemaphore.release();
                return null;
            });
        } else {
            if (log.isDebugEnabled()) {
                log.debug("[{}] Failed Partition-Metadata lookup due to too many lookup-requests {}", remoteAddress,
                        topicName);
            }
            commandSender.sendPartitionMetadataResponse(ServerError.TooManyRequests,
                    "Failed due to too many pending lookup requests", requestId);
        }
    }
@Test(expectedExceptions = IllegalArgumentException.class)
    public void shouldFailHandlePartitionMetadataRequest() throws Exception {
        ServerCnx serverCnx = mock(ServerCnx.class, CALLS_REAL_METHODS);
        Field stateUpdater = ServerCnx.class.getDeclaredField("state");
        stateUpdater.setAccessible(true);
        stateUpdater.set(serverCnx, ServerCnx.State.Failed);
        serverCnx.handlePartitionMetadataRequest(any());
    }