public DescribeTopicPartitionsResponseData handleDescribeTopicPartitionsRequest(RequestChannel.Request abstractRequest) {
        DescribeTopicPartitionsRequestData request = ((DescribeTopicPartitionsRequest) abstractRequest.loggableRequest()).data();
        Set<String> topics = new HashSet<>();
        boolean fetchAllTopics = request.topics().isEmpty();
        DescribeTopicPartitionsRequestData.Cursor cursor = request.cursor();
        String cursorTopicName = cursor != null ? cursor.topicName() : "";
        if (fetchAllTopics) {
            JavaConverters.asJavaCollection(metadataCache.getAllTopics()).forEach(topicName -> {
                if (topicName.compareTo(cursorTopicName) >= 0) {
                    topics.add(topicName);
                }
            });
        } else {
            request.topics().forEach(topic -> {
                String topicName = topic.name();
                if (topicName.compareTo(cursorTopicName) >= 0) {
                    topics.add(topicName);
                }
            });

            if (cursor != null && !topics.contains(cursor.topicName())) {
                // The topic in cursor must be included in the topic list if provided.
                throw new InvalidRequestException("DescribeTopicPartitionsRequest topic list should contain the cursor topic: " + cursor.topicName());
            }
        }

        if (cursor != null && cursor.partitionIndex() < 0) {
            // The partition id in cursor must be valid.
            throw new InvalidRequestException("DescribeTopicPartitionsRequest cursor partition must be valid: " + cursor);
        }

        // Do not disclose the existence of topics unauthorized for Describe, so we've not even checked if they exist or not
        Set<DescribeTopicPartitionsResponseTopic> unauthorizedForDescribeTopicMetadata = new HashSet<>();

        Stream<String> authorizedTopicsStream = topics.stream().sorted().filter(topicName -> {
            boolean isAuthorized = authHelper.authorize(
                abstractRequest.context(), DESCRIBE, TOPIC, topicName, true, true, 1);
            if (!fetchAllTopics && !isAuthorized) {
                // We should not return topicId when on unauthorized error, so we return zero uuid.
                unauthorizedForDescribeTopicMetadata.add(describeTopicPartitionsResponseTopic(
                    Errors.TOPIC_AUTHORIZATION_FAILED, topicName, Uuid.ZERO_UUID, false, Collections.emptyList())
                );
            }
            return isAuthorized;
        });

        DescribeTopicPartitionsResponseData response = metadataCache.getTopicMetadataForDescribeTopicResponse(
            JavaConverters.asScalaIterator(authorizedTopicsStream.iterator()),
            abstractRequest.context().listenerName,
            (String topicName) -> topicName.equals(cursorTopicName) ? cursor.partitionIndex() : 0,
            Math.max(Math.min(config.maxRequestPartitionSizeLimit(), request.responsePartitionLimit()), 1),
            fetchAllTopics
        );

        // get topic authorized operations
        response.topics().forEach(topicData ->
            topicData.setTopicAuthorizedOperations(authHelper.authorizedOperations(abstractRequest, new Resource(TOPIC, topicData.name()))));

        response.topics().addAll(unauthorizedForDescribeTopicMetadata);
        return response;
    }
@Test
    void testDescribeTopicPartitionsRequestWithEdgeCases() {
        // 1. Set up authorizer
        Authorizer authorizer = mock(Authorizer.class);
        String authorizedTopic = "authorized-topic1";
        String authorizedTopic2 = "authorized-topic2";

        Action expectedActions1 = new Action(AclOperation.DESCRIBE, new ResourcePattern(ResourceType.TOPIC, authorizedTopic, PatternType.LITERAL), 1, true, true);
        Action expectedActions2 = new Action(AclOperation.DESCRIBE, new ResourcePattern(ResourceType.TOPIC, authorizedTopic2, PatternType.LITERAL), 1, true, true);

        when(authorizer.authorize(any(RequestContext.class), argThat(t ->
            t.contains(expectedActions1) || t.contains(expectedActions2))))
            .thenAnswer(invocation -> {
                List<Action> actions = invocation.getArgument(1);
                return actions.stream().map(action -> {
                    if (action.resourcePattern().name().startsWith("authorized"))
                        return AuthorizationResult.ALLOWED;
                    else
                        return AuthorizationResult.DENIED;
                }).collect(Collectors.toList());
            });

        // 2. Set up MetadataCache
        Uuid authorizedTopicId = Uuid.randomUuid();
        Uuid authorizedTopicId2 = Uuid.randomUuid();

        Map<String, Uuid> topicIds = new HashMap<>();
        topicIds.put(authorizedTopic, authorizedTopicId);
        topicIds.put(authorizedTopic2, authorizedTopicId2);

        BrokerEndpointCollection collection = new BrokerEndpointCollection();
        collection.add(new BrokerEndpoint()
                .setName(broker.endpoints().get(0).listener())
                .setHost(broker.endpoints().get(0).host())
                .setPort(broker.endpoints().get(0).port())
                .setSecurityProtocol(broker.endpoints().get(0).securityProtocol())
        );
        List<ApiMessage> records = Arrays.asList(
            new RegisterBrokerRecord()
                .setBrokerId(broker.id())
                .setBrokerEpoch(0)
                .setIncarnationId(Uuid.randomUuid())
                .setEndPoints(collection)
                .setRack(broker.rack())
                .setFenced(false),
            new TopicRecord().setName(authorizedTopic).setTopicId(topicIds.get(authorizedTopic)),
            new TopicRecord().setName(authorizedTopic2).setTopicId(topicIds.get(authorizedTopic2)),
            new PartitionRecord()
                .setTopicId(authorizedTopicId)
                .setPartitionId(0)
                .setReplicas(Arrays.asList(0, 1, 2))
                .setLeader(0)
                .setIsr(Arrays.asList(0))
                .setEligibleLeaderReplicas(Arrays.asList(1))
                .setLastKnownElr(Arrays.asList(2))
                .setLeaderEpoch(0)
                .setPartitionEpoch(1)
                .setLeaderRecoveryState(LeaderRecoveryState.RECOVERED.value()),
            new PartitionRecord()
                .setTopicId(authorizedTopicId)
                .setPartitionId(1)
                .setReplicas(Arrays.asList(0, 1, 2))
                .setLeader(0)
                .setIsr(Arrays.asList(0))
                .setEligibleLeaderReplicas(Arrays.asList(1))
                .setLastKnownElr(Arrays.asList(2))
                .setLeaderEpoch(0)
                .setPartitionEpoch(1)
                .setLeaderRecoveryState(LeaderRecoveryState.RECOVERED.value()),
            new PartitionRecord()
                .setTopicId(authorizedTopicId2)
                .setPartitionId(0)
                .setReplicas(Arrays.asList(0, 1, 3))
                .setLeader(0)
                .setIsr(Arrays.asList(0))
                .setEligibleLeaderReplicas(Arrays.asList(1))
                .setLastKnownElr(Arrays.asList(3))
                .setLeaderEpoch(0)
                .setPartitionEpoch(2)
                .setLeaderRecoveryState(LeaderRecoveryState.RECOVERED.value())
        );
        KRaftMetadataCache metadataCache = new KRaftMetadataCache(0, () -> KRaftVersion.KRAFT_VERSION_1);
        updateKraftMetadataCache(metadataCache, records);
        DescribeTopicPartitionsRequestHandler handler =
            new DescribeTopicPartitionsRequestHandler(metadataCache, new AuthHelper(scala.Option.apply(authorizer)), createKafkaDefaultConfig());

        // 3.1 With cursor point to the first one
        DescribeTopicPartitionsRequest describeTopicPartitionsRequest = new DescribeTopicPartitionsRequest(new DescribeTopicPartitionsRequestData()
            .setTopics(Arrays.asList(
                new DescribeTopicPartitionsRequestData.TopicRequest().setName(authorizedTopic),
                new DescribeTopicPartitionsRequestData.TopicRequest().setName(authorizedTopic2)
                ))
            .setCursor(new DescribeTopicPartitionsRequestData.Cursor().setTopicName(authorizedTopic).setPartitionIndex(1))
        );

        RequestChannel.Request request;
        try {
            request = buildRequest(describeTopicPartitionsRequest, plaintextListener);
        } catch (Exception e) {
            fail(e.getMessage());
            return;
        }
        DescribeTopicPartitionsResponseData response = handler.handleDescribeTopicPartitionsRequest(request);
        List<DescribeTopicPartitionsResponseTopic> topics = response.topics().valuesList();
        assertEquals(2, topics.size());
        DescribeTopicPartitionsResponseTopic topicToCheck = topics.get(0);
        assertEquals(authorizedTopicId, topicToCheck.topicId());
        assertEquals(Errors.NONE.code(), topicToCheck.errorCode());
        assertEquals(authorizedTopic, topicToCheck.name());
        assertEquals(1, topicToCheck.partitions().size());

        topicToCheck = topics.get(1);
        assertEquals(authorizedTopicId2, topicToCheck.topicId());
        assertEquals(Errors.NONE.code(), topicToCheck.errorCode());
        assertEquals(authorizedTopic2, topicToCheck.name());
        assertEquals(1, topicToCheck.partitions().size());

        // 3.2 With cursor point to the second one. The first topic should be ignored.
        describeTopicPartitionsRequest = new DescribeTopicPartitionsRequest(new DescribeTopicPartitionsRequestData()
            .setTopics(Arrays.asList(
                new DescribeTopicPartitionsRequestData.TopicRequest().setName(authorizedTopic),
                new DescribeTopicPartitionsRequestData.TopicRequest().setName(authorizedTopic2)
                ))
            .setCursor(new DescribeTopicPartitionsRequestData.Cursor().setTopicName(authorizedTopic2).setPartitionIndex(0))
        );

        try {
            request = buildRequest(describeTopicPartitionsRequest, plaintextListener);
        } catch (Exception e) {
            fail(e.getMessage());
            return;
        }
        response = handler.handleDescribeTopicPartitionsRequest(request);
        topics = response.topics().valuesList();
        assertEquals(1, topics.size());
        topicToCheck = topics.get(0);
        assertEquals(authorizedTopicId2, topicToCheck.topicId());
        assertEquals(Errors.NONE.code(), topicToCheck.errorCode());
        assertEquals(authorizedTopic2, topicToCheck.name());
        assertEquals(1, topicToCheck.partitions().size());

        // 3.3 With cursor point to a non existing topic. Exception should be thrown if not querying all the topics.
        describeTopicPartitionsRequest = new DescribeTopicPartitionsRequest(new DescribeTopicPartitionsRequestData()
            .setTopics(Arrays.asList(
                new DescribeTopicPartitionsRequestData.TopicRequest().setName(authorizedTopic),
                new DescribeTopicPartitionsRequestData.TopicRequest().setName(authorizedTopic2)
                ))
            .setCursor(new DescribeTopicPartitionsRequestData.Cursor().setTopicName("Non-existing").setPartitionIndex(0))
        );

        try {
            handler.handleDescribeTopicPartitionsRequest(buildRequest(describeTopicPartitionsRequest, plaintextListener));
        } catch (Exception e) {
            assertInstanceOf(InvalidRequestException.class, e, e.getMessage());
        }

        // 3.4 With cursor point to a negative partition id. Exception should be thrown if not querying all the topics.
        describeTopicPartitionsRequest = new DescribeTopicPartitionsRequest(new DescribeTopicPartitionsRequestData()
            .setTopics(Arrays.asList(
                new DescribeTopicPartitionsRequestData.TopicRequest().setName(authorizedTopic),
                new DescribeTopicPartitionsRequestData.TopicRequest().setName(authorizedTopic2)
            ))
            .setCursor(new DescribeTopicPartitionsRequestData.Cursor().setTopicName(authorizedTopic).setPartitionIndex(-1))
        );

        try {
            handler.handleDescribeTopicPartitionsRequest(buildRequest(describeTopicPartitionsRequest, plaintextListener));
        } catch (Exception e) {
            assertInstanceOf(InvalidRequestException.class, e, e.getMessage());
        }
    }