@Override
    public ProduceResponse getErrorResponse(int throttleTimeMs, Throwable e) {
        /* In case the producer doesn't actually want any response */
        if (acks == 0) return null;
        ApiError apiError = ApiError.fromThrowable(e);
        ProduceResponseData data = new ProduceResponseData().setThrottleTimeMs(throttleTimeMs);
        partitionSizes().forEach((tp, ignored) -> {
            ProduceResponseData.TopicProduceResponse tpr = data.responses().find(tp.topic());
            if (tpr == null) {
                tpr = new ProduceResponseData.TopicProduceResponse().setName(tp.topic());
                data.responses().add(tpr);
            }
            tpr.partitionResponses().add(new ProduceResponseData.PartitionProduceResponse()
                    .setIndex(tp.partition())
                    .setRecordErrors(Collections.emptyList())
                    .setBaseOffset(INVALID_OFFSET)
                    .setLogAppendTimeMs(RecordBatch.NO_TIMESTAMP)
                    .setLogStartOffset(INVALID_OFFSET)
                    .setErrorMessage(apiError.message())
                    .setErrorCode(apiError.error().code()));
        });
        return new ProduceResponse(data);
    }
@return data or IllegalStateException if the data is removed (to prevent unnecessary memory retention).
clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java