Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/producer/internals/ProduceRequestResult.java (Java)

Code before:
    /**
     * Set the result of the produce request.
     *
     * @param baseOffset The base offset assigned to the record
     * @param logAppendTime The log append time or -1 if CreateTime is being used
     * @param errorsByIndex Function mapping the batch index to the exception, or null if the response was successful
     */
    public void set(long baseOffset, long logAppendTime, Function<Integer, RuntimeException> errorsByIndex) {
        this.baseOffset = baseOffset;
        this.logAppendTime = logAppendTime;
        this.errorsByIndex = errorsByIndex;
    }

    /**
     * Mark this request as complete and unblock any threads waiting on its completion.
     */
    public void done() {
        if (baseOffset == null)
            throw new IllegalStateException("The method `set` must be invoked before this method.");
        this.latch.countDown();
    }

    /**
     * Add a dependent ProduceRequestResult that must complete before this result is considered complete.
     * This is used when a batch is split into multiple batches - the original batch's result
     * should not complete until all split batches have completed.
     *
     * @param dependentResult The dependent result to wait for
     */
    public void addDependentResult(ProduceRequestResult dependentResult) {
        synchronized (dependentResults) {
            dependentResults.add(dependentResult);
        }
    }

    /**
     * Await the completion of this request.
     *
     * This only waits for THIS request's latch and not dependent results.
     * When a batch is split into multiple batches, dependent results are created and tracked
     * separately, but this method does not wait for them. Individual record futures automatically
     * handle waiting for their respective split batch via {@link FutureRecordMetadata#chain(FutureRecordMetadata)},
     * which redirects the future to point to the correct split batch's result.
     *
     * For flush() semantics that require waiting for all dependent results, use
     * {@link #awaitAllDependents()}.
     */
    public void await() throws InterruptedException {
        latch.await();
    }


Review comment (performance):
>  the original batch's result should not complete until all split batches have completed.

In some cases like flush(), the original batch's result should not complete until all split batches have completed.


Code after:
     *
     * @param topicPartition The topic and partition to which this record set was sent
     */
    public ProduceRequestResult(TopicPartition topicPartition) {
        this.topicPartition = topicPartition;
    }

    /**
     * Set the result of the produce request.
     *
     * @param baseOffset The base offset assigned to the record
     * @param logAppendTime The log append time or -1 if CreateTime is being used
     * @param errorsByIndex Function mapping the batch index to the exception, or null if the response was successful
     */
    public void set(long baseOffset, long logAppendTime, Function<Integer, RuntimeException> errorsByIndex) {
        this.baseOffset = baseOffset;
        this.logAppendTime = logAppendTime;
        this.errorsByIndex = errorsByIndex;
    }

    /**
     * Mark this request as complete and unblock any threads waiting on its completion.
     */
    public void done() {
        if (baseOffset == null)
            throw new IllegalStateException("The method `set` must be invoked before this method.");
        this.latch.countDown();
    }

    /**
     * Add a dependent ProduceRequestResult.
     * This is used when a batch is split into multiple batches - in some cases like flush(), the original
     * batch's result should not complete until all split batches have completed.
     *
     * @param dependentResult The dependent result to wait for
     */
    public void addDependent(ProduceRequestResult dependentResult) {
        synchronized (dependentResults) {
            dependentResults.add(dependentResult);
        }
    }

    /**
     * Await the completion of this request.
     *
     * This only waits for THIS request's latch and not dependent results.
     * When a batch is split into multiple batches, dependent results are created and tracked
     * separately, but this method does not wait for them. Individual record futures automatically
     * handle waiting for their respective split batch via {@link FutureRecordMetadata#chain(FutureRecordMetadata)},
     * which redirects the future to point to the correct split batch's result.
     *