@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 testInvalidResponse() {
        AbortTransactionHandler handler = new AbortTransactionHandler(abortSpec, logContext);

        WriteTxnMarkersResponseData response = new WriteTxnMarkersResponseData();
        assertFailed(KafkaException.class, topicPartition, handler.handleResponse(node, singleton(topicPartition),
            new WriteTxnMarkersResponse(response)));

        WriteTxnMarkersResponseData.WritableTxnMarkerResult markerResponse =
            new WriteTxnMarkersResponseData.WritableTxnMarkerResult();
        response.markers().add(markerResponse);
        assertFailed(KafkaException.class, topicPartition, handler.handleResponse(node, singleton(topicPartition),
            new WriteTxnMarkersResponse(response)));

        markerResponse.setProducerId(abortSpec.producerId());
        assertFailed(KafkaException.class, topicPartition, handler.handleResponse(node, singleton(topicPartition),
            new WriteTxnMarkersResponse(response)));

        WriteTxnMarkersResponseData.WritableTxnMarkerTopicResult topicResponse =
            new WriteTxnMarkersResponseData.WritableTxnMarkerTopicResult();
        markerResponse.topics().add(topicResponse);
        assertFailed(KafkaException.class, topicPartition, handler.handleResponse(node, singleton(topicPartition),
            new WriteTxnMarkersResponse(response)));

        topicResponse.setName(abortSpec.topicPartition().topic());
        assertFailed(KafkaException.class, topicPartition, handler.handleResponse(node, singleton(topicPartition),
            new WriteTxnMarkersResponse(response)));

        WriteTxnMarkersResponseData.WritableTxnMarkerPartitionResult partitionResponse =
            new WriteTxnMarkersResponseData.WritableTxnMarkerPartitionResult();
        topicResponse.partitions().add(partitionResponse);
        assertFailed(KafkaException.class, topicPartition, handler.handleResponse(node, singleton(topicPartition),
            new WriteTxnMarkersResponse(response)));

        partitionResponse.setPartitionIndex(abortSpec.topicPartition().partition());
        topicResponse.setName(abortSpec.topicPartition().topic() + "random");
        assertFailed(KafkaException.class, topicPartition, handler.handleResponse(node, singleton(topicPartition),
            new WriteTxnMarkersResponse(response)));

        topicResponse.setName(abortSpec.topicPartition().topic());
        markerResponse.setProducerId(abortSpec.producerId() + 1);
        assertFailed(KafkaException.class, topicPartition, handler.handleResponse(node, singleton(topicPartition),
            new WriteTxnMarkersResponse(response)));
    }