artemlivshits commented on code in PR #12570:
URL: https://github.com/apache/kafka/pull/12570#discussion_r970291640
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java:
##########
@@ -1129,31 +1129,34 @@ public void onCompletion(RecordMetadata metadata,
Exception exception) {
assertEquals(1, mockRandom.get());
// Produce large record, we should exceed "sticky" limit, but
produce to this partition
- // as we switch after the "sticky" limit is exceeded. The
partition is switched after
- // we produce.
+ // as we try to switch after the "sticky" limit is exceeded. The
switch is disabled
+ // because of incomplete batch.
Review Comment:
Right now we check the isFull condition, which seems to be false if the
batch is not closed and there is one record in the batch -- when a new batch is
allocated it is allocated to accommodate the new record, so it can be larger
than batch.size (if the first record is greater than batch.size) making the
isFull check false. Here is the relevant code
Allocation:
```
int size = Math.max(this.batchSize,
AbstractRecords.estimateSizeInBytesUpperBound(maxUsableMagic, compression, key,
value, headers));
log.trace("Allocating a new {} byte message buffer for
topic {} partition {} with remaining timeout {}ms", size, topic, partition,
maxTimeToBlock);
// This call may block if we exhausted buffer space.
buffer = free.allocate(size, maxTimeToBlock);
```
Check:
```
public boolean isFull() {
// note that the write limit is respected only after the first
record is added which ensures we can always
// create non-empty batches (this is used to disable batching when
the producer's batch size is set to 0).
return appendStream == CLOSED_STREAM || (this.numRecords > 0 &&
this.writeLimit <= estimatedBytesWritten());
}
```
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
