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

Code before:
        boolean adjustSequenceNumbers,
        boolean deallocateBatch
    ) {
        this.sensors.recordErrors(batch.topicPartition.topic(), batch.recordCount);

        if (batch.completeExceptionally(topLevelException, recordExceptions)) {
            if (transactionManager != null) {
                try {
                    // This call can throw an exception in the rare case that there's an invalid state transition
                    // attempted. Catch these so as not to interfere with the rest of the logic.
                    transactionManager.handleFailedBatch(batch, topLevelException, adjustSequenceNumbers);
                } catch (Exception e) {
                    log.debug("Encountered error when transaction manager was handling a failed batch", e);
                }
            }
            if (deallocateBatch) {
                maybeRemoveAndDeallocateBatch(batch);
            } else {
                // Fix for KAFKA-19012
                // The pooled ByteBuffer associated with this batch might still be in use by the network client so we
                // cannot allow it to be reused yet. We skip deallocating it now, instead doing that in completeBatch
                // or in another run of failBatch where we hit the else clause and call deallocateAlreadyRemovedIncomplete
                maybeRemoveAndDeallocateBatchLater(batch);
            }
        } else {
            this.accumulator.deallocate(batch);
        }
    }

    /**
     * We can retry a send if the error is transient and the number of attempts taken is fewer than the maximum allowed.
     * We can also retry OutOfOrderSequence exceptions for future batches, since if the first batch has failed, the
     * future batches are certain to fail with an OutOfOrderSequence exception.
     */
    private boolean canRetry(ProducerBatch batch, ProduceResponse.PartitionResponse response, long now) {
        return !batch.hasReachedDeliveryTimeout(accumulator.getDeliveryTimeoutMs(), now) &&
            batch.attempts() < this.retries &&
            !batch.isDone() &&
            (transactionManager == null ?
                    response.error.exception() instanceof RetriableException :
                    transactionManager.canRetry(response, batch));
    }

    /**
     * Transfer the record batches into a list of produce requests on a per-node basis
     */
    private void sendProduceRequests(Map<Integer, List<ProducerBatch>> collated, long now) {
        for (Map.Entry<Integer, List<ProducerBatch>> entry : collated.entrySet())
            sendProduceRequest(now, entry.getKey(), acks, requestTimeoutMs, entry.getValue());
    }


Review comment (bug):
> We skip deallocating it now, instead doing that in completeBatch
>  or in another run of failBatch where we hit the else clause and call deallocateAlreadyRemovedIncomplete

How about "We skip deallocating it now. When the request in network client completes with a response, either completeBatch() or failBatch() will be called with deallocateBatch=true. The buffer associated with the batch will be deallocated then."?

Code after:
        RuntimeException topLevelException,
        Function<Integer, RuntimeException> recordExceptions,
        boolean adjustSequenceNumbers,
        boolean deallocateBatch
    ) {
        this.sensors.recordErrors(batch.topicPartition.topic(), batch.recordCount);

        if (batch.completeExceptionally(topLevelException, recordExceptions)) {
            if (transactionManager != null) {
                try {
                    // This call can throw an exception in the rare case that there's an invalid state transition
                    // attempted. Catch these so as not to interfere with the rest of the logic.
                    transactionManager.handleFailedBatch(batch, topLevelException, adjustSequenceNumbers);
                } catch (Exception e) {
                    log.debug("Encountered error when transaction manager was handling a failed batch", e);
                }
            }
            if (deallocateBatch) {
                maybeRemoveAndDeallocateBatch(batch);
            } else {
                // Fix for KAFKA-19012
                // The pooled ByteBuffer associated with this batch might still be in use by the network client so we
                // cannot allow it to be reused yet. We skip deallocating it now. When the request in the network client 
                // completes with a response, either completeBatch() or failBatch() will be called with deallocateBatch=true.
                // The buffer associated with the batch will be deallocated then.
                maybeRemoveAndDeallocateBatchLater(batch);
            }
        } else {
            if (deallocateBatch) {
                this.accumulator.deallocate(batch);
            }
        }
    }

    /**
     * We can retry a send if the error is transient and the number of attempts taken is fewer than the maximum allowed.
     * We can also retry OutOfOrderSequence exceptions for future batches, since if the first batch has failed, the
     * future batches are certain to fail with an OutOfOrderSequence exception.
     */
    private boolean canRetry(ProducerBatch batch, ProduceResponse.PartitionResponse response, long now) {
        return !batch.hasReachedDeliveryTimeout(accumulator.getDeliveryTimeoutMs(), now) &&
            batch.attempts() < this.retries &&
            !batch.isDone() &&
            (transactionManager == null ?
                    response.error.exception() instanceof RetriableException :
                    transactionManager.canRetry(response, batch));
    }

    /**
     * Transfer the record batches into a list of produce requests on a per-node basis
     */