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

Code before:

    /**
     * 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.
     * Note: This only waits for the local latch and not dependent results.
     * The dependent results are tracked for completion status via completed() method,
     * but individual record futures handle chaining via FutureRecordMetadata.chain().
     */
    public void await() throws InterruptedException {
        latch.await();
    }

    /**
     * Await the completion of this request (up to the given time interval)
     * @param timeout The maximum time to wait
     * @param unit The unit for the max time
     * @return true if the request completed, false if we timed out
     */
    public boolean await(long timeout, TimeUnit unit) throws InterruptedException {
        return latch.await(timeout, unit);
    }

    /**
     * The base offset for the request (the first offset in the record set)
     */
    public long baseOffset() {
        return baseOffset;
    }

    /**

Review comment (performance):
flush() actually calls await() and it needs to wait for all dependent requests to complete. Perhaps we could add a separate method like awaitAllDependants() and use that in flush().
```
    public void awaitFlushCompletion() throws InterruptedException {
        try {
            // Obtain a copy of all of the incomplete ProduceRequestResult(s) at the time of the flush.
            // We must be careful not to hold a reference to the ProduceBatch(s) so that garbage
            // collection can occur on the contents.
            // The sender will remove ProducerBatch(s) from the original incomplete collection.
            for (ProduceRequestResult result : this.incomplete.requestResults())
                result.await();
        } finally {
            this.flushesInProgress.decrementAndGet();
        }
    }
```

Code after:
     * @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.
     *
     * For flush() semantics that require waiting for all dependent results, use
     * {@link #awaitAllDependents()}.
     */
    public void await() throws InterruptedException {
        latch.await();
    }

    /**
     * Await the completion of this request (up to the given time interval)
     * @param timeout The maximum time to wait