@Override
    public WriteTxnMarkersRequest.Builder buildBatchedRequest(
        int brokerId,
        Set<TopicPartition> topicPartitions
    ) {
        validateTopicPartitions(topicPartitions);

        WriteTxnMarkersRequestData.WritableTxnMarker marker = new WriteTxnMarkersRequestData.WritableTxnMarker()
            .setCoordinatorEpoch(abortSpec.coordinatorEpoch())
            .setProducerEpoch(abortSpec.producerEpoch())
            .setProducerId(abortSpec.producerId())
            .setTransactionResult(false);

        marker.topics().add(new WriteTxnMarkersRequestData.WritableTxnMarkerTopic()
            .setName(abortSpec.topicPartition().topic())
            .setPartitionIndexes(singletonList(abortSpec.topicPartition().partition()))
        );

        WriteTxnMarkersRequestData request = new WriteTxnMarkersRequestData();
        request.markers().add(marker);

        return new WriteTxnMarkersRequest.Builder(request);
    }
@Test
    public void testValidBuildRequestCall() {
        AbortTransactionHandler handler = new AbortTransactionHandler(abortSpec, logContext);
        WriteTxnMarkersRequest.Builder request = handler.buildBatchedRequest(1, singleton(topicPartition));
        assertEquals(1, request.data.markers().size());

        WriteTxnMarkersRequestData.WritableTxnMarker markerRequest = request.data.markers().get(0);
        assertEquals(abortSpec.producerId(), markerRequest.producerId());
        assertEquals(abortSpec.producerEpoch(), markerRequest.producerEpoch());
        assertEquals(abortSpec.coordinatorEpoch(), markerRequest.coordinatorEpoch());
        assertEquals(1, markerRequest.topics().size());

        WriteTxnMarkersRequestData.WritableTxnMarkerTopic topicRequest = markerRequest.topics().get(0);
        assertEquals(abortSpec.topicPartition().topic(), topicRequest.name());
        assertEquals(singletonList(abortSpec.topicPartition().partition()), topicRequest.partitionIndexes());
    }