@Override
    public ApiResult<TopicPartition, Void> handleResponse(
        Node broker,
        Set<TopicPartition> topicPartitions,
        AbstractResponse abstractResponse
    ) {
        validateTopicPartitions(topicPartitions);

        WriteTxnMarkersResponse response = (WriteTxnMarkersResponse) abstractResponse;
        List<WriteTxnMarkersResponseData.WritableTxnMarkerResult> markerResponses = response.data().markers();

        if (markerResponses.size() != 1 || markerResponses.get(0).producerId() != abortSpec.producerId()) {
            return ApiResult.failed(abortSpec.topicPartition(), new KafkaException("WriteTxnMarkers response " +
                "included unexpected marker entries: " + markerResponses + "(expected to find exactly one " +
                "entry with producerId " + abortSpec.producerId() + ")"));
        }

        WriteTxnMarkersResponseData.WritableTxnMarkerResult markerResponse = markerResponses.get(0);
        List<WriteTxnMarkersResponseData.WritableTxnMarkerTopicResult> topicResponses = markerResponse.topics();

        if (topicResponses.size() != 1 || !topicResponses.get(0).name().equals(abortSpec.topicPartition().topic())) {
            return ApiResult.failed(abortSpec.topicPartition(), new KafkaException("WriteTxnMarkers response " +
                "included unexpected topic entries: " + markerResponses + "(expected to find exactly one " +
                "entry with topic partition " + abortSpec.topicPartition() + ")"));
        }

        WriteTxnMarkersResponseData.WritableTxnMarkerTopicResult topicResponse = topicResponses.get(0);
        List<WriteTxnMarkersResponseData.WritableTxnMarkerPartitionResult> partitionResponses =
            topicResponse.partitions();

        if (partitionResponses.size() != 1 || partitionResponses.get(0).partitionIndex() != abortSpec.topicPartition().partition()) {
            return ApiResult.failed(abortSpec.topicPartition(), new KafkaException("WriteTxnMarkers response " +
                "included unexpected partition entries for topic " + abortSpec.topicPartition().topic() +
                ": " + markerResponses + "(expected to find exactly one entry with partition " +
                abortSpec.topicPartition().partition() + ")"));
        }

        WriteTxnMarkersResponseData.WritableTxnMarkerPartitionResult partitionResponse = partitionResponses.get(0);
        Errors error = Errors.forCode(partitionResponse.errorCode());

        if (error != Errors.NONE) {
            return handleError(error);
        } else {
            return ApiResult.completed(abortSpec.topicPartition(), null);
        }
    }
@Test
    public void testInvalidHandleResponseCall() {
        AbortTransactionHandler handler = new AbortTransactionHandler(abortSpec, logContext);
        WriteTxnMarkersResponseData response = new WriteTxnMarkersResponseData();
        assertThrows(IllegalArgumentException.class, () -> handler.handleResponse(node,
            emptySet(), new WriteTxnMarkersResponse(response)));
        assertThrows(IllegalArgumentException.class, () -> handler.handleResponse(node,
            mkSet(new TopicPartition("foo", 1)), new WriteTxnMarkersResponse(response)));
        assertThrows(IllegalArgumentException.class, () -> handler.handleResponse(node,
            mkSet(topicPartition, new TopicPartition("foo", 1)), new WriteTxnMarkersResponse(response)));
    }