@Override
    public ApiResult<TopicPartition, DeletedRecords> handleResponse(
        Node broker,
        Set<TopicPartition> keys,
        AbstractResponse abstractResponse
    ) {
        DeleteRecordsResponse response = (DeleteRecordsResponse) abstractResponse;
        Map<TopicPartition, DeletedRecords> completed = new HashMap<>();
        Map<TopicPartition, Throwable> failed = new HashMap<>();
        List<TopicPartition> unmapped = new ArrayList<>();
        Set<TopicPartition> retriable = new HashSet<>();

        for (DeleteRecordsResponseData.DeleteRecordsTopicResult topicResult: response.data().topics()) {
            for (DeleteRecordsResponseData.DeleteRecordsPartitionResult partitionResult : topicResult.partitions()) {
                Errors error = Errors.forCode(partitionResult.errorCode());
                TopicPartition topicPartition = new TopicPartition(topicResult.name(), partitionResult.partitionIndex());
                if (error == Errors.NONE) {
                    completed.put(topicPartition, new DeletedRecords(partitionResult.lowWatermark()));
                } else {
                    handlePartitionError(topicPartition, error, failed, unmapped, retriable);
                }
            }
        }

        // Sanity-check if the current leader for these partitions returned results for all of them
        for (TopicPartition topicPartition : keys) {
            if (unmapped.isEmpty()
                    && !completed.containsKey(topicPartition)
                    && !failed.containsKey(topicPartition)
                    && !retriable.contains(topicPartition)
            ) {
                ApiException sanityCheckException = new ApiException(
                        "The response from broker " + broker.id() +
                                " did not contain a result for topic partition " + topicPartition);
                log.error(
                        "DeleteRecords request for topic partition {} failed sanity check",
                        topicPartition,
                        sanityCheckException);
                failed.put(topicPartition, sanityCheckException);
            }
        }

        return new ApiResult<>(completed, failed, unmapped);
    }
@Test
    public void testMixedResponse() {
        Map<TopicPartition, Short> errorsByPartition = new HashMap<>();

        TopicPartition errorPartition = t0p0;
        Errors error = Errors.UNKNOWN_SERVER_ERROR;
        errorsByPartition.put(errorPartition, error.code());

        TopicPartition retriableErrorPartition = t0p1;
        Errors retriableError = Errors.NOT_LEADER_OR_FOLLOWER;
        errorsByPartition.put(retriableErrorPartition, retriableError.code());

        TopicPartition retriableErrorPartition2 = t0p2;
        Errors retriableError2 = Errors.REQUEST_TIMED_OUT;
        errorsByPartition.put(retriableErrorPartition2, retriableError2.code());

        AdminApiHandler.ApiResult<TopicPartition, DeletedRecords> result =
                handleResponse(createResponse(errorsByPartition));

        Set<TopicPartition> completed = new HashSet<>(recordsToDelete.keySet());

        Map<TopicPartition, Throwable> failed = new HashMap<>();
        failed.put(errorPartition, error.exception());
        completed.removeAll(failed.keySet());

        List<TopicPartition> unmapped = new ArrayList<>();
        unmapped.add(retriableErrorPartition);
        completed.removeAll(unmapped);

        Set<TopicPartition> retriable = singleton(retriableErrorPartition2);
        completed.removeAll(retriable);

        assertResult(result, completed, failed, unmapped, retriable);
    }