chia7712 commented on code in PR #21065:
URL: https://github.com/apache/kafka/pull/21065#discussion_r2665898730
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java:
##########
@@ -1031,12 +1031,30 @@ BuiltInPartitioner createBuiltInPartitioner(LogContext
logContext, String topic,
*/
public void deallocate(ProducerBatch batch) {
incomplete.remove(batch);
+ deallocateIfNecessary(batch);
+ }
+
+ private void deallocateIfNecessary(ProducerBatch batch) {
// Only deallocate the batch if it is not a split batch because split
batch are allocated outside the
// buffer pool.
- if (!batch.isSplitBatch())
+ if (!batch.isSplitBatch() && !batch.isBufferDeallocated() &&
batch.setBufferDeallocated(true))
Review Comment:
WDYT about simplifying this logic like this?
```java
public boolean tryMarkDeallocated() {
return this.bufferDeallocated.compareAndSet(false, true);
}
private void deallocateIfNecessary(ProducerBatch batch) {
// Only deallocate the batch if it is not a split batch because split
batch are allocated outside the
// buffer pool.
if (!batch.isSplitBatch() && batch.tryMarkDeallocated())
free.deallocate(batch.buffer(), batch.initialCapacity());
}
```
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -749,12 +759,16 @@ private void completeBatch(ProducerBatch batch,
ProduceResponse.PartitionRespons
if (batch.complete(response.baseOffset, response.logAppendTime)) {
maybeRemoveAndDeallocateBatch(batch);
+ } else {
+ // Always safe to call dellocate because the batch keeps track of
whether or not it was deallocated yet
+ this.accumulator.deallocateAlreadyRemovedIncomplete(batch);
}
}
private void failBatch(ProducerBatch batch,
ProduceResponse.PartitionResponse response,
- boolean adjustSequenceNumbers) {
+ boolean adjustSequenceNumbers,
+ boolean deallocateBatch) {
Review Comment:
It appears the input is always `true`, so this argument might be redundant?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
