@Override
    public WriteTxnMarkersResponse getErrorResponse(int throttleTimeMs, Throwable e) {
        Errors error = Errors.forException(e);

        final Map<Long, Map<TopicPartition, Errors>> errors = new HashMap<>(data.markers().size());
        for (WritableTxnMarker markerEntry : data.markers()) {
            Map<TopicPartition, Errors> errorsPerPartition = new HashMap<>();
            for (WritableTxnMarkerTopic topic : markerEntry.topics()) {
                for (Integer partitionIdx : topic.partitionIndexes()) {
                    errorsPerPartition.put(new TopicPartition(topic.name(), partitionIdx), error);
                }
            }
            errors.put(markerEntry.producerId(), errorsPerPartition);
        }

        return new WriteTxnMarkersResponse(errors);
    }
@Test
    public void testGetErrorResponse() {
        WriteTxnMarkersRequest.Builder builder = new WriteTxnMarkersRequest.Builder(ApiKeys.WRITE_TXN_MARKERS.latestVersion(), markers);
        for (short version : ApiKeys.WRITE_TXN_MARKERS.allVersions()) {
            WriteTxnMarkersRequest request = builder.build(version);
            WriteTxnMarkersResponse errorResponse =
                request.getErrorResponse(throttleTimeMs, Errors.UNKNOWN_PRODUCER_ID.exception());

            assertEquals(Collections.singletonMap(
                TOPIC_PARTITION, Errors.UNKNOWN_PRODUCER_ID), errorResponse.errorsByProducerId().get(PRODUCER_ID));
            assertEquals(Collections.singletonMap(Errors.UNKNOWN_PRODUCER_ID, 1), errorResponse.errorCounts());
            // Write txn marker has no throttle time defined in response.
            assertEquals(0, errorResponse.throttleTimeMs());
        }
    }