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

Code before:
     * The error thrown (generally on the server) while processing this request
     */
    public RuntimeException error(int batchIndex) {
        if (errorsByIndex == null) {
            return null;
        } else {
            return errorsByIndex.apply(batchIndex);
        }
    }

    /**
     * The topic and partition to which the record was appended
     */
    public TopicPartition topicPartition() {
        return topicPartition;
    }

    /**
     * Has the request completed?
     */
    public boolean completed() {
        if (this.latch.getCount() != 0L) {
            return false;
        }
        // are all the dependent results completed?
        synchronized (dependentResults) {
            for (ProduceRequestResult dependentResult : dependentResults) {
                if (!dependentResult.completed()) {
                    return false;
                }
            }
        }
        return true;
    }
}


Review comment (performance):
Hmm, this is not quite right. This method is called from FutureRecordMetadata. If a batch is split into two batches b1 and b2. For records in b1, they don't need to wait for the completion for b2. So, it doesn't seem that we should change this method. But it will be useful to document the semantic that it doesn't wait for dependent requests.
 
```
    @Override
    public boolean isDone() {
        if (nextRecordMetadata != null)
            return nextRecordMetadata.isDone();
        return this.result.completed();
    }

```

Code after:
     * @throws InterruptedException if the thread is interrupted while waiting
     */
    public void awaitAllDependents() throws InterruptedException {
        Queue<ProduceRequestResult> toWait = new ArrayDeque<>();
        toWait.add(this);

        while (!toWait.isEmpty()) {
            ProduceRequestResult current = toWait.poll();

            // first wait for THIS result's latch to be released
            current.latch.await();

            // add all dependent split batches to the queue.
            // we synchronize to get a consistent snapshot, then release the lock
            // before continuing but the actual waiting happens outside the lock.
            synchronized (current.dependentResults) {
                toWait.addAll(current.dependentResults);
            }
        }
    }

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

    /**
     * Return true if log append time is being used for this topic
     */
    public boolean hasLogAppendTime() {
        return logAppendTime != RecordBatch.NO_TIMESTAMP;
    }

    /**
     * The log append time or -1 if CreateTime is being used
     */
    public long logAppendTime() {
        return logAppendTime;
    }

    /**
     * The error thrown (generally on the server) while processing this request
     */
    public RuntimeException error(int batchIndex) {
        if (errorsByIndex == null) {
            return null;
        } else {
            return errorsByIndex.apply(batchIndex);
        }